Serialization and rejection.

src/supervisor/dispatchers.py

@@ -226,6 +226,7 @@ class PEventListenerDispatcher(PDispatcher):
         return False
     
     def readable(self):
+        self.handle_listener_state_change()
         return True
 
     def handle_read_event(self):
@@ -309,12 +310,14 @@ class PEventListenerDispatcher(PDispatcher):
                 tokenlen = len(self.EVENT_REJECTED_TOKEN)
                 self.state_buffer = self.state_buffer[tokenlen:]
                 process.listener_state = EventListenerStates.ACKNOWLEDGED
+                notify(EventRejectedEvent(process, process.event))
                 process.event = None
             else:
                 msg = '%s: BUSY -> UNKNOWN' % procname
                 self._trace(msg)
                 process.listener_state = EventListenerStates.UNKNOWN
                 self.state_buffer = ''
+                notify(EventRejectedEvent(process, process.event))
                 process.event = None
             if self.state_buffer:
                 # keep going til its too short

src/supervisor/options.py

@@ -1518,15 +1518,14 @@ class EventListenerConfig(ProcessConfig):
         dispatchers = {}
         from supervisor.dispatchers import PEventListenerDispatcher
         from supervisor.dispatchers import PInputDispatcher
+        from supervisor.dispatchers import POutputDispatcher
         from supervisor import events
         if stdout_fd is not None:
-            etype = events.ProcessCommunicationStdoutEvent
             dispatchers[stdout_fd] = PEventListenerDispatcher(proc, 'stdout',
                                                               stdout_fd)
         if stderr_fd is not None:
             etype = events.ProcessCommunicationStderrEvent
-            dispatchers[stderr_fd] = PEventListenerDispatcher(proc, 'stderr',
-                                                              stderr_fd)
+            dispatchers[stderr_fd] = POutputDispatcher(proc, etype, stderr_fd)
         if stdin_fd is not None:
             dispatchers[stdin_fd] = PInputDispatcher(proc, 'stdin', stdin_fd)
         return dispatchers, p

src/supervisor/process.py

@@ -564,10 +564,12 @@ class EventListenerPool(ProcessGroupBase):
                                                '%s' % event)
                 self.event_buffer.insert(0, event)
 
-    def _eventEnvelope(self, event, payload):
+    def _eventEnvelope(self, event_type, payload):
+        event_name = getEventNameByType(event_type)
+        payload_len = len(payload)
         D = {'ver':'3.0',
-             'len':len(payload),
-             'event_name':getEventNameByType(event),
+             'len':payload_len,
+             'event_name':event_name,
              'payload':payload}
         return 'SUPERVISORD%(ver)s %(event_name)s %(len)s\n%(payload)s' % D
 
@@ -585,6 +587,7 @@ class EventListenerPool(ProcessGroupBase):
     def _dispatchEvent(self, event, buffer=True):
         # events are required to be instances
         serializer = None
+        event_type = event.__class__
         for klass, callback in serializers.items():
             if isinstance(event, klass):
                 serializer = callback
@@ -596,7 +599,8 @@ class EventListenerPool(ProcessGroupBase):
             if process.listener_state == EventListenerStates.READY:
                 payload = serializer(event)
                 try:
-                    process.write(self._eventEnvelope(event, payload))
+                    envelope = self._eventEnvelope(event_type, payload)
+                    process.write(envelope)
                 except IOError, why:
                     if why[0] == errno.EPIPE:
                         continue
@@ -623,7 +627,8 @@ def overflow_event(event):
 serializers[events.EventBufferOverflowEvent] = overflow_event
 
 def proc_sc_event(event):
-    return ''
+    return 'process_name: %s\n' % event.process.config.name
+
 serializers[events.ProcessStateChangeEvent] = proc_sc_event
 
 def supervisor_sc_event(event):