def __init__(self, config, state=None):
        self.config = config
        self.logsremoved = False
        self.stop_called = False
        self.stop_report_called = True
        self.backoff_secs = None
        self.spawned = False
        if state is None:
            from supervisor.process import ProcessStates

            state = ProcessStates.RUNNING
        self.state = state
        self.error_at_clear = False
        self.killed_with = None
        self.drained = False
        self.stdout_buffer = as_bytes("")
        self.stderr_buffer = as_bytes("")
        self.stdout_logged = as_bytes("")
        self.stderr_logged = as_bytes("")
        self.stdin_buffer = as_bytes("")
        self.pipes = {}
        self.rpipes = {}
        self.dispatchers = {}
        self.finished = None
        self.logs_reopened = False
        self.execv_arg_exception = None
        self.input_fd_drained = None
        self.output_fd_drained = None
        self.transitioned = False
        self.write_error = None

    def test_handle_more_follow_file_recreated(self):
        request = DummyRequest('/logtail/foo', None, None, None)
        tmpfile = TempFileOpen()
        try:
            tmpfile.write(as_bytes('a' * 80))
            tmpfile.flush()

            producer = self._makeOne(request, tmpfile.name, 80)
            result = producer.more()

            self.assertEqual(result, as_bytes('a' * 80))

            # in windows open files need to be closed
            tmpfile.close()

            tmpfile = open(tmpfile.name, 'wb')
            try:
                tmpfile.write(as_bytes('b' * 80))
                tmpfile.flush()
                result = producer.more()
            finally:
                tmpfile.close()
                os.remove(tmpfile.name)
        finally:
            tmpfile.close()
        self.assertEqual(result, as_bytes('b' * 80))

    def refill_buffer (self):
        """ Implement deferreds """
        while 1:
            if len(self.producer_fifo):
                p = self.producer_fifo.first()
                # a 'None' in the producer fifo is a sentinel,
                # telling us to close the channel.
                if p is None:
                    if not self.ac_out_buffer:
                        self.producer_fifo.pop()
                        self.close()
                    return
                elif isinstance(p, str):
                    self.producer_fifo.pop()
                    self.ac_out_buffer += p
                    return

                data = p.more()

                if data is NOT_DONE_YET:
                    self.delay = p.delay
                    return

                elif data:
                    try:
                        self.ac_out_buffer = self.ac_out_buffer + data
                    except TypeError:
                        self.ac_out_buffer = as_bytes(self.ac_out_buffer) + as_bytes(data)

                    self.delay = False
                    return
                else:
                    self.producer_fifo.pop()
            else:
                return

 def more (self):
     while len(self.buffer) < self.buffer_size:
         data = self.producer.more()
         if data is NOT_DONE_YET:
             return NOT_DONE_YET
         if data:
             try:
                 self.buffer = self.buffer + data
             except TypeError:
                 self.buffer = as_bytes(self.buffer) + as_bytes(data)
         else:
             break
     r = self.buffer
     self.buffer = ''
     return r

    def handle_request(self, request):
        # authorize a request before handling it...
        scheme = get_header(AUTHORIZATION, request.header)

        if scheme:
            scheme = scheme.lower()
            if scheme == "basic":
                cookie = get_header(AUTHORIZATION, request.header, 2)
                try:
                    decoded = as_string(decodestring(as_bytes(cookie)))
                except:
                    print_function("malformed authorization info <%s>" % cookie)
                    request.error(400)
                    return
                auth_info = decoded.split(":", 1)
                if self.authorizer.authorize(auth_info):
                    self.pass_count.increment()
                    request.auth_info = auth_info
                    self.handler.handle_request(request)
                else:
                    self.handle_unauthorized(request)
            # elif scheme == 'digest':
            #       print 'digest: ',AUTHORIZATION.group(2)
            else:
                print("unknown/unsupported auth method: %s" % scheme)
                self.handle_unauthorized(request)
        else:
            # list both?  prefer one or the other?
            # you could also use a 'nonce' here. [see below]
            # auth = 'Basic realm="%s" Digest realm="%s"' % (self.realm, self.realm)
            # nonce = self.make_nonce (request)
            # auth = 'Digest realm="%s" nonce="%s"' % (self.realm, nonce)
            # request['WWW-Authenticate'] = auth
            # print 'sending header: %s' % request['WWW-Authenticate']
            self.handle_unauthorized(request)

    def _dispatchEvent(self, event):
        pool_serial = event.pool_serials[self.config.name]

        for process in self.processes.values():
            if process.state != ProcessStates.RUNNING:
                continue
            if process.listener_state == EventListenerStates.READY:
                payload = str(event)
                try:
                    event_type = event.__class__
                    serial = event.serial
                    envelope = self._eventEnvelope(event_type, serial,
                                                   pool_serial, payload)
                    process.write(as_bytes(envelope))
                except OSError as why:
                    if why.args[0] != errno.EPIPE:
                        raise
                    continue

                process.listener_state = EventListenerStates.BUSY
                process.event = event
                self.config.options.logger.debug(
                    'event %s sent to listener %s' % (
                    event.serial, process.config.name))
                return True

        return False

    def request(self, host, handler, request_body, verbose=0):
        if not self.connection:
            self.connection = self._get_connection()
            self.headers = {
                "User-Agent" : self.user_agent,
                "Content-Type" : "text/xml",
                "Accept": "text/xml"
                }

            # basic auth
            if self.username is not None and self.password is not None:
                unencoded = "%s:%s" % (self.username, self.password)
                encoded = as_string(encodestring(as_bytes(unencoded)))
                encoded = encoded.replace('\n', '')
                encoded = encoded.replace('\012', '')
                self.headers["Authorization"] = "Basic %s" % encoded

        self.headers["Content-Length"] = str(len(request_body))

        self.connection.request('POST', handler, request_body, self.headers)

        r = self.connection.getresponse()

        if r.status != 200:
            self.connection.close()
            self.connection = None
            raise xmlrpclib.ProtocolError(host + handler,
                                          r.status,
                                          r.reason,
                                          '' )
        data = r.read()
        p, u = self.getparser()
        p.feed(data)
        p.close()
        return u.close()

 def test_handle_more_follow_file_recreated(self):
     request = DummyRequest('/logtail/foo', None, None, None)
     f = tempfile.NamedTemporaryFile()
     f.write(as_bytes('a' * 80))
     f.flush()
     producer = self._makeOne(request, f.name, 80)
     result = producer.more()
     self.assertEqual(result, as_string('a' * 80))
     f.close()
     f2 = open(f.name, 'wb')
     try:
         f2.write(as_bytes('b' * 80))
         f2.close()
         result = producer.more()
     finally:
         os.unlink(f2.name)
     self.assertEqual(result, as_string('b' * 80))

 def test_handle_request_does_not_authorize_bad_credentials(self):
     request = DummyRequest('/logtail/process1', None, None, None)
     encoded = base64.b64encode(as_bytes("wrong:wrong"))
     request.header = ["Authorization: Basic %s" % as_string(encoded)]
     handler = DummyHandler()
     auth_handler = self._makeOne({'user':'password'}, handler)
     auth_handler.handle_request(request)
     self.assertFalse(handler.handled_request)

 def push (self, thing):
     # Sometimes, text gets pushed by XMLRPC logic for later
     # processing.
     if isinstance(thing, str):
         thing = as_bytes(thing)
     if isinstance(thing, bytes):
         thing = producers.simple_producer(thing, buffer_size=len(thing))
     self.outgoing.append(thing)

 def pull_trigger(self, thunk=None):
     # print 'PULL_TRIGGER: ', len(self.thunks)
     if thunk:
         self.lock.acquire()
         try:
             self.thunks.append(thunk)
         finally:
             self.lock.release()
     os.write(self.trigger, as_bytes("x"))

 def __call__(self, *args, **kwargs):
     message = args[-1]
     if sys.version_info < (3, 0) and isinstance(message, unicode):
         # Python 2.x raises a UnicodeEncodeError when attempting to
         #  transmit unicode characters that don't encode in the
         #  default encoding.
         args = list(args)
         args[-1] = as_bytes(message)
     super(MockSysLog, self).__call__(*args, **kwargs)

 def test_handle_more_fd_closed(self):
     request = DummyRequest('/logtail/foo', None, None, None)
     with tempfile.NamedTemporaryFile() as f:
         f.write(as_bytes('a' * 80))
         f.flush()
         producer = self._makeOne(request, f.name, 80)
         producer.file.close()
         result = producer.more()
     self.assertEqual(result, producer.more())

 def test_handle_more_follow_file_gone(self):
     request = DummyRequest('/logtail/foo', None, None, None)
     filename = tempfile.mktemp()
     with open(filename, 'wb') as f:
         f.write(as_bytes('a' * 80))
     try:
         producer = self._makeOne(request, f.name, 80)
     finally:
         os.unlink(f.name)
     result = producer.more()
     self.assertEqual(result, as_string('a' * 80))
     with open(filename, 'wb') as f:
         f.write(as_bytes('b' * 80))
     try:
         result = producer.more() # should open in new file
         self.assertEqual(result, as_string('b' * 80))
     finally:
          os.unlink(f.name)

 def test_handle_request_authorizes_good_password_with_colon(self):
     request = DummyRequest('/logtail/process1', None, None, None)
     # password contains colon
     encoded = base64.b64encode(as_bytes("user:pass:word"))
     request.header = ["Authorization: Basic %s" % as_string(encoded)]
     handler = DummyHandler()
     auth_handler = self._makeOne({'user':'pass:word'}, handler)
     auth_handler.handle_request(request)
     self.assertTrue(handler.handled_request)

 def test_handle_more(self):
     request = DummyRequest('/logtail/foo', None, None, None)
     from supervisor import http
     f = tempfile.NamedTemporaryFile()
     f.write(as_bytes('a' * 80))
     f.flush()
     producer = self._makeOne(request, f.name, 80)
     result = producer.more()
     self.assertEqual(result, as_string('a' * 80))
     f.write(as_bytes('w' * 100))
     f.flush()
     result = producer.more()
     self.assertEqual(result, as_string('w' * 100))
     result = producer.more()
     self.assertEqual(result, http.NOT_DONE_YET)
     f.truncate(0)
     f.flush()
     result = producer.more()
     self.assertEqual(result, '==> File truncated <==\n')

 def handle_write_event(self):
     if self.input_buffer:
         try:
             self.flush()
         except OSError as why:
             if why.args[0] == errno.EPIPE:
                 self.input_buffer = as_bytes('')
                 self.close()
             else:
                 raise

    def test_handle_more_follow_file_gone(self):
        request = DummyRequest('/logtail/foo', None, None, None)
        tmpfile = TempFileOpen()
        tmpfile.write(as_bytes('a' * 80))
        try:
            producer = self._makeOne(request, tmpfile.name, 80)
            producer.close()
        finally:
            tmpfile.close()
        result = producer.more()
        self.assertEqual(result, '')

        with open(tmpfile.name, 'wb') as tmpfile:
            tmpfile.write(as_bytes('b' * 80))
        try:
            result = producer.more()  # should open in new file
            self.assertEqual(result, as_bytes('b' * 80))
        finally:
            os.remove(tmpfile.name)

 def test_handle_more(self):
     request = DummyRequest('/logtail/foo', None, None, None)
     from supervisor import http
     f = tempfile.NamedTemporaryFile()
     f.write(as_bytes('a' * 80))
     f.flush()
     t = f.name
     producer = self._makeOne(request, t, 80)
     result = producer.more()
     self.assertEqual(result, TAILF_LOG_WRAPPER.format(cgi.escape(as_bytes('a' * 80).decode('utf-8'))))
     f.write(as_bytes('w' * 100))
     f.flush()
     result = producer.more()
     self.assertEqual(result, TAILF_LOG_WRAPPER.format(cgi.escape(as_bytes('w' * 100).decode('utf-8'))))
     result = producer.more()
     self.assertEqual(result, http.NOT_DONE_YET)
     f.truncate(0)
     f.flush()
     result = producer.more()
     self.assertEqual(result, '==> File truncated <==\n')

 def authorize(self, auth_info):
     username, password = auth_info
     if username in self.dict:
         stored_password = self.dict[username]
         if stored_password.startswith('{SHA}'):
             password_hash = sha1(as_bytes(password)).hexdigest()
             return stored_password[5:] == password_hash
         else:
             return stored_password == password
     else:
         return False