# Path: duct/utils.py
# def wait(msecs):
#     def __init__(self, react, path, **kwargs):
#     def _getEndpoint(self, *_a):
# def reverseNameFromIPAddress(address):
#     def __init__(self):
#     def reverse(self, ip):
#         def _ret(result, ip):
#     def __init__(self, finished):
#     def dataReceived(self, data):
#     def connectionLost(self, *_r):
#     def __init__(self, body):
#     def startProducing(self, consumer):
#     def pauseProducing(self):
#     def stopProducing(self):
#     def __init__(self, deferred, timeout):
#     def outReceived(self, data):
#     def errReceived(self, data):
#     def processEnded(self, reason):
#     def connectionMade(self):
#         def killIfAlive():
# def fork(executable, args=(), env={}, path=None, timeout=3600):
#         def getContext(self, *_a):
#     def __init__(self, timeout=120):
#     def abort_request(self, request):
#     def response(self, request):
#     def request(self, url, method='GET', headers={}, data=None, socket=None,
#                 follow_redirect=False):
#             def timeoutProxy(request):
#             def requestAborted(failure):
#     def getBody(self, url, method='GET', headers={}, data=None, socket=None):
#     def getJson(self, url, method='GET', headers={}, data=None, socket=None):
#     def __init__(self, location='/var/lib/duct/cache'):
#     def _changed(self):
#     def _acquire_cache(self):
#     def _write_cache(self, data):
#     def _persist(self):
#     def _read(self):
#     def _remove_key(self, key):
#     def expire(self, age):
#     def set(self, key, val):
#     def get(self, k):
#     def contains(self, k):
#     def delete(self, k):
# class SocketyAgent(Agent):
# class Timeout(Exception):
# class Resolver(object):
# class BodyReceiver(protocol.Protocol):
# class StringProducer(object):
# class ProcessProtocol(protocol.ProcessProtocol):
#     class WebClientContextFactory(ClientContextFactory):
# class HTTPRequest(object):
# class PersistentCache(object):
#     SSL = True
#     SSL = False


import time
import uuid
import json
from base64 import b64encode
from duct import utils
"""
.. module:: elasticsearch
   :synopsis: Elasticsearch protocol module

.. moduleauthor:: Colin Alston <colin@imcol.in>
"""



class ElasticSearch(object):
    """Twisted ElasticSearch API
    """
    def __init__(self, url='http://localhost:9200', user=None, password=None,
                 index='duct-%Y.%m.%d'):
        self.url = url.rstrip('/')
        self.index = index
        self.user = user
        self.password = password

    def _get_index(self):
        return time.strftime(self.index)

    def _request(self, path, data=None, method='GET'):
        headers = {}
        if self.user:
            authorization = b64encode('%s:%s' % (self.user, self.password)
                                     ).decode()
            headers['Authorization'] = ['Basic ' + authorization]


# Path: duct/utils.py
# def wait(msecs):
#     def __init__(self, react, path, **kwargs):
#     def _getEndpoint(self, *_a):
# def reverseNameFromIPAddress(address):
#     def __init__(self):
#     def reverse(self, ip):
#         def _ret(result, ip):
#     def __init__(self, finished):
#     def dataReceived(self, data):
#     def connectionLost(self, *_r):
#     def __init__(self, body):
#     def startProducing(self, consumer):
#     def pauseProducing(self):
#     def stopProducing(self):
#     def __init__(self, deferred, timeout):
#     def outReceived(self, data):
#     def errReceived(self, data):
#     def processEnded(self, reason):
#     def connectionMade(self):
#         def killIfAlive():
# def fork(executable, args=(), env={}, path=None, timeout=3600):
#         def getContext(self, *_a):
#     def __init__(self, timeout=120):
#     def abort_request(self, request):
#     def response(self, request):
#     def request(self, url, method='GET', headers={}, data=None, socket=None,
#                 follow_redirect=False):
#             def timeoutProxy(request):
#             def requestAborted(failure):
#     def getBody(self, url, method='GET', headers={}, data=None, socket=None):
#     def getJson(self, url, method='GET', headers={}, data=None, socket=None):
#     def __init__(self, location='/var/lib/duct/cache'):
#     def _changed(self):
#     def _acquire_cache(self):
#     def _write_cache(self, data):
#     def _persist(self):
#     def _read(self):
#     def _remove_key(self, key):
#     def expire(self, age):
#     def set(self, key, val):
#     def get(self, k):
#     def contains(self, k):
#     def delete(self, k):
# class SocketyAgent(Agent):
# class Timeout(Exception):
# class Resolver(object):
# class BodyReceiver(protocol.Protocol):
# class StringProducer(object):
# class ProcessProtocol(protocol.ProcessProtocol):
#     class WebClientContextFactory(ClientContextFactory):
# class HTTPRequest(object):
# class PersistentCache(object):
#     SSL = True
#     SSL = False


# Path: duct/protocol/elasticsearch.py
import time
import uuid
import json
from base64 import b64encode
from duct import utils

"""
.. module:: elasticsearch
   :synopsis: Elasticsearch protocol module

.. moduleauthor:: Colin Alston <colin@imcol.in>
"""



class ElasticSearch(object):
    """Twisted ElasticSearch API
    """
    def __init__(self, url='http://localhost:9200', user=None, password=None,
                 index='duct-%Y.%m.%d'):
        self.url = url.rstrip('/')
        self.index = index
        self.user = user
        self.password = password

    def _get_index(self):
        return time.strftime(self.index)

    def _request(self, path, data=None, method='GET'):
        headers = {}
        if self.user:
            authorization = b64encode('%s:%s' % (self.user, self.password)
                                     ).decode()
            headers['Authorization'] = ['Basic ' + authorization]


        return utils.HTTPRequest().getJson(