Use __str__ on events instead of separate serializers.

src/supervisor/events.py

@@ -39,14 +39,26 @@ class EventBufferOverflowEvent(Event):
         self.group = group
         self.event = event
 
+    def __str__(self):
+        name = self.group.config.name
+        typ = getEventNameByType(self.event)
+        return 'group_name: %s\nevent_type: %s' % (name, typ)
+
 class ProcessCommunicationEvent(Event):
     # event mode tokens
     BEGIN_TOKEN = '<!--XSUPERVISOR:BEGIN-->'
     END_TOKEN   = '<!--XSUPERVISOR:END-->'
+    channel = None # this is abstract
     def __init__(self, process, data):
         self.process = process
         self.data = data
 
+    def __str__(self):
+        return 'process_name: %s\nchannel: %s\n%s' % (
+            self.process.config.name,
+            self.channel,
+            self.data)
+
 class ProcessCommunicationStdoutEvent(ProcessCommunicationEvent):
     channel = 'stdout'
 
@@ -60,6 +72,10 @@ class ProcessStateChangeEvent(Event):
     def __init__(self, process):
         self.process = process
 
+    def __str__(self):
+        return 'process_name: %s' % self.process.config.name
+
+
 class StartingFromStoppedEvent(ProcessStateChangeEvent):
     frm = ProcessStates.STOPPED
     to = ProcessStates.STARTING
@@ -117,6 +133,8 @@ class ToUnknownEvent(ProcessStateChangeEvent):
 
 class SupervisorStateChangeEvent(Event):
     """ Abstract class """
+    def __str__(self):
+        return ''
 
 class SupervisorRunningEvent(SupervisorStateChangeEvent):
     pass
@@ -185,3 +203,4 @@ def getProcessStateChangeEventType(old_state, new_state):
     return typ
         
         
+

src/supervisor/process.py

@@ -565,9 +565,10 @@ class EventListenerPool(ProcessGroupBase):
             event = self.event_buffer.pop(0)
             ok = self._dispatchEvent(event, buffer=False)
             if not ok:
-                self.config.options.logger.log(self.config.options.TRACE,
-                                               'Failed sending buffered event '
-                                               '%s' % event.serial)
+                self.config.options.logger.log(
+                    self.config.options.TRACE,
+                    'Failed sending buffered event %s (bufsize %s)' % (
+                    event.serial, len(self.event_buffer)))
                 self.event_buffer.insert(0, event)
 
     def _eventEnvelope(self, event_type, serial, payload):
@@ -597,22 +598,14 @@ 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
-        if serializer is None:
-            # this is a system programming error, we must handle
-            # all events
-            raise NotImplementedError(etype)
         if not hasattr(event, 'serial'):
             event.serial = new_serial()
         for process in self.processes.values():
             if process.state != ProcessStates.RUNNING:
                 continue
             if process.listener_state == EventListenerStates.READY:
-                payload = serializer(event)
+                payload = str(event)
                 try:
                     serial = event.serial
                     envelope = self._eventEnvelope(event_type, serial, payload)
@@ -626,7 +619,8 @@ class EventListenerPool(ProcessGroupBase):
                 process.event = event
                 self.config.options.logger.log(
                     self.config.options.TRACE,
-                    'event %s sent to listener' % event.serial)
+                    'event %s sent to listener %s' % (
+                    event.serial, process.config.name))
                 return True
         if buffer:
             self._bufferEvent(event)
@@ -640,28 +634,5 @@ def new_serial():
     _num = _num + 1
     return val
 
-serializers = {}
-def pcomm_event(event):
-    return 'process_name: %s\nchannel: %s\n%s' % (
-        event.process.config.name,
-        event.channel,
-        event.data)
-serializers[events.ProcessCommunicationEvent] = pcomm_event
-
-def overflow_event(event):
-    name = event.group.config.name
-    typ = getEventNameByType(event.event)
-    return 'group_name: %s\nevent_type: %s' % (name, typ)
-serializers[events.EventBufferOverflowEvent] = overflow_event
-
-def proc_sc_event(event):
-    return 'process_name: %s' % event.process.config.name
-
-serializers[events.ProcessStateChangeEvent] = proc_sc_event
-
-def supervisor_sc_event(event):
-    return ''
-serializers[events.SupervisorStateChangeEvent] = supervisor_sc_event
-
             
     

src/supervisor/tests/test_events.py

@@ -1,6 +1,10 @@
 import sys
 import unittest
 
+from supervisor.tests.base import DummyOptions
+from supervisor.tests.base import DummyPConfig
+from supervisor.tests.base import DummyProcess
+
 class EventSubscriptionNotificationTests(unittest.TestCase):
     def setUp(self):
         from supervisor import events
@@ -91,6 +95,49 @@ class TestEventTypes(unittest.TestCase):
         inst = ProcessStateChangeEvent(1)
         self.assertEqual(inst.process, 1)
         
+class TestSerializations(unittest.TestCase):
+    def test_pcomm_stdout_event(self):
+        options = DummyOptions()
+        pconfig1 = DummyPConfig(options, 'process1', 'process1','/bin/process1')
+        process1 = DummyProcess(pconfig1)
+        from supervisor.events import ProcessCommunicationStdoutEvent
+        event = ProcessCommunicationStdoutEvent(process1, 'yo')
+        self.assertEqual(str(event),
+                         'process_name: process1\nchannel: stdout\nyo'
+                         )
+            
+    def test_pcomm_stderr_event(self):
+        options = DummyOptions()
+        pconfig1 = DummyPConfig(options, 'process1', 'process1','/bin/process1')
+        process1 = DummyProcess(pconfig1)
+        from supervisor.events import ProcessCommunicationStderrEvent
+        event = ProcessCommunicationStderrEvent(process1, 'yo')
+        self.assertEqual(str(event),
+                         'process_name: process1\nchannel: stderr\nyo'
+                         )
+
+    def test_overflow_event(self):
+        from supervisor import events
+        options = DummyOptions()
+        pconfig1 = DummyPConfig(options, 'foo', 'process1','/bin/process1')
+        process1 = DummyProcess(pconfig1)
+        wrapped = events.ProcessCommunicationStderrEvent(process1, 'yo')
+        event = events.EventBufferOverflowEvent(process1, wrapped)
+        self.assertEqual(str(event), 'group_name: foo\nevent_type: None')
+
+    def test_process_sc_event(self):
+        from supervisor import events
+        options = DummyOptions()
+        pconfig1 = DummyPConfig(options, 'process1', 'process1','/bin/process1')
+        process1 = DummyProcess(pconfig1)
+        event = events.StartingFromStoppedEvent(process1)
+        self.assertEqual(str(event), 'process_name: process1')
+
+    def test_supervisor_sc_event(self):
+        from supervisor import events
+        event = events.SupervisorRunningEvent()
+        self.assertEqual(str(event), '')
+
 class TestUtilityFunctions(unittest.TestCase):
     def test_getEventNameByType(self):
         from supervisor import events

src/supervisor/tests/test_process.py

@@ -1127,52 +1127,6 @@ class EventListenerPoolTests(ProcessGroupBaseTests):
         self.assertEqual(process1.listener_state, EventListenerStates.BUSY)
         self.assertEqual(process1.event, event)
 
-class TestSerializers(unittest.TestCase):
-    def test_pcomm_event(self):
-        options = DummyOptions()
-        pconfig1 = DummyPConfig(options, 'process1', 'process1','/bin/process1')
-        process1 = DummyProcess(pconfig1)
-        class DummyPCommEvent:
-            process = process1
-            channel = 'stdout'
-            data = 'yo'
-        event = DummyPCommEvent()
-        from supervisor.process import pcomm_event
-        self.assertEqual(pcomm_event(event),
-                         'process_name: process1\nchannel: stdout\nyo'
-                         )
-            
-    def test_overflow_event(self):
-        class DummyConfig:
-            name = 'foo'
-        class DummyGroup:
-            config = DummyConfig()
-        from supervisor.events import StartingFromStoppedEvent
-        class DummyOverflowEvent:
-            group = DummyGroup()
-            event = StartingFromStoppedEvent(None)
-        event = DummyOverflowEvent()
-        from supervisor.process import overflow_event
-        self.assertEqual(overflow_event(event),
-                         'group_name: foo\nevent_type: None')
-
-    def test_pcomm_event(self):
-        options = DummyOptions()
-        pconfig1 = DummyPConfig(options, 'process1', 'process1','/bin/process1')
-        process1 = DummyProcess(pconfig1)
-        class DummyStateChangeEvent:
-            process = process1
-        event = DummyStateChangeEvent()
-        from supervisor.process import proc_sc_event
-        self.assertEqual(proc_sc_event(event), 'process_name: process1')
-
-    def test_supervisor_sc_event(self):
-        class DummyEvent:
-            pass
-        event = DummyEvent()
-        from supervisor.process import supervisor_sc_event
-        self.assertEqual(supervisor_sc_event(event), '')
-
 def test_suite():
     return unittest.findTestCases(sys.modules[__name__])