Make the parser recognize eventlistener grocess groups (although currently they do little different than normal process groups do).

src/supervisor/events.py

@@ -30,3 +30,7 @@ class ProcessCommunicationEvent:
         self.process_name = process_name
         self.channel = channel
         self.data = data
+
+EVENT_NAMES = {
+    'PROCESS_COMMUNICATION_EVENT':ProcessCommunicationEvent,
+    }

src/supervisor/options.py

@@ -708,7 +708,7 @@ class ServerOptions(Options):
                 ProcessGroupConfig(self, group_name, priority, group_processes)
                 )
 
-        # process homogeneous groups
+        # process "normal" homogeneous groups
         for section in all_sections:
             if ( (not section.startswith('program:') )
                  or section in homogeneous_exclude ):
@@ -720,6 +720,31 @@ class ServerOptions(Options):
                 ProcessGroupConfig(self, program_name, priority, processes)
                 )
 
+        # process "event listener" homogeneous groups
+        for section in all_sections:
+            if not section.startswith('eventlistener:'):
+                 continue
+            pool_name = section.split(':', 1)[1]
+            priority = integer(get(section, 'priority', 999))
+            pool_event_names = [x.upper() for x in
+                                list_of_strings(get(section, 'events', ''))]
+            if not pool_event_names:
+                raise ValueError('[%s] section requires an "events" line' %
+                                 section)
+            from supervisor.events import EVENT_NAMES
+            pool_events = []
+            for pool_event_name in pool_event_names:
+                pool_event = EVENT_NAMES.get(pool_event_name)
+                if pool_event is None:
+                    raise ValueError('Unknown event type %s in [%s] events' %
+                                     (pool_event_name, section))
+                pool_events.append(pool_event)
+            processes=self.processes_from_section(parser, section, pool_name)
+            groups.append(
+                EventListenerPoolConfig(self, pool_name, priority, processes,
+                                        pool_events)
+                )
+
         groups.sort()
         return groups
 
@@ -1161,10 +1186,6 @@ class ServerOptions(Options):
         from supervisor.process import Subprocess
         return Subprocess(config)
 
-    def make_group(self, config):
-        from supervisor.process import ProcessGroup
-        return ProcessGroup(config)
-
     def make_pipes(self, stderr=True):
         """ Create pipes for parent to child stdin/stdout/stderr
         communications.  Open fd in nonblocking mode so we can read them
@@ -1494,6 +1515,26 @@ class ProcessGroupConfig(Config):
         for config in self.process_configs:
             config.create_autochildlogs()
 
+    def make_group(self):
+        from supervisor.process import ProcessGroup
+        return ProcessGroup(self)
+
+class EventListenerPoolConfig(Config):
+    def __init__(self, options, name, priority, process_configs, pool_events):
+        self.options = options
+        self.name = name
+        self.priority = priority
+        self.process_configs = process_configs
+        self.pool_events = pool_events
+
+    def after_setuid(self):
+        for config in self.process_configs:
+            config.create_autochildlogs()
+
+    def make_group(self):
+        from supervisor.process import ProcessGroup
+        return ProcessGroup(self)
+
 class BasicAuthTransport(xmlrpclib.Transport):
     """ A transport that understands basic auth and UNIX domain socket
     URLs """

src/supervisor/supervisord.py

@@ -100,7 +100,7 @@ class Supervisor:
         try:
             for config in self.options.process_group_configs:
                 name = config.name
-                self.process_groups[name] = self.options.make_group(config)
+                self.process_groups[name] = config.make_group()
             self.options.process_environment()
             self.options.openhttpserver(self)
             self.options.setsignals()

src/supervisor/tests/base.py

@@ -114,9 +114,6 @@ class DummyOptions:
     def make_process(self, config):
         return DummyProcess(config)
 
-    def make_group(self, config):
-        return DummyProcessGroup(config)
-
     def kill(self, pid, sig):
         if self.kill_error:
             raise OSError(self.kill_error)
@@ -651,6 +648,9 @@ class DummyPGroupConfig:
     def after_setuid(self):
         self.after_setuid_called = True
 
+    def make_group(self):
+        return DummyProcessGroup(self)
+
 class DummyProcessGroup:
     def __init__(self, config):
         self.config = config

src/supervisor/tests/test_options.py

@@ -9,6 +9,7 @@ import signal
 
 from supervisor.tests.base import DummyLogger
 from supervisor.tests.base import DummyOptions
+from supervisor.tests.base import DummyPConfig
 from supervisor.tests.base import lstrip
 
 class ServerOptionsTests(unittest.TestCase):
@@ -439,7 +440,69 @@ class ServerOptionsTests(unittest.TestCase):
         self.assertEqual(gconfig.name, 'many')
         self.assertEqual(gconfig.priority, 1)
         self.assertEqual(len(gconfig.process_configs), 2)
-        
+
+    def test_event_listener_pools_from_parser(self):
+        text = lstrip("""\
+        [eventlistener:dog]
+        events=PROCESS_COMMUNICATION_EVENT
+        process_name = %(program_name)s_%(process_num)s
+        command = /bin/dog
+        numprocs = 2
+        priority = 1
+
+        [eventlistener:cat]
+        events=PROCESS_COMMUNICATION_EVENT
+        process_name = %(program_name)s_%(process_num)s
+        command = /bin/cat
+        numprocs = 3
+        priority = 1
+        """)
+        from supervisor.options import UnhosedConfigParser
+        config = UnhosedConfigParser()
+        config.read_string(text)
+        instance = self._makeOne()
+        gconfigs = instance.process_groups_from_parser(config)
+        self.assertEqual(len(gconfigs), 2)
+
+        gconfig1 = gconfigs[0]
+        self.assertEqual(gconfig1.name, 'dog')
+        self.assertEqual(gconfig1.priority, 1)
+        self.assertEqual(len(gconfig1.process_configs), 2)
+
+        gconfig1 = gconfigs[1]
+        self.assertEqual(gconfig1.name, 'cat')
+        self.assertEqual(gconfig1.priority, 1)
+        self.assertEqual(len(gconfig1.process_configs), 3)
+
+    def test_event_listener_pool_noeventsline(self):
+        text = lstrip("""\
+        [eventlistener:dog]
+        process_name = %(program_name)s_%(process_num)s
+        command = /bin/dog
+        numprocs = 2
+        priority = 1
+        """)
+        from supervisor.options import UnhosedConfigParser
+        config = UnhosedConfigParser()
+        config.read_string(text)
+        instance = self._makeOne()
+        self.assertRaises(ValueError,instance.process_groups_from_parser,config)
+
+    def test_event_listener_pool_unknown_eventtype(self):
+        text = lstrip("""\
+        [eventlistener:dog]
+        events=PROCESS_COMMUNICATION_EVENT,THIS_EVENT_TYPE_DOESNT_EXIST
+        process_name = %(program_name)s_%(process_num)s
+        command = /bin/dog
+        numprocs = 2
+        priority = 1
+        """)
+        from supervisor.options import UnhosedConfigParser
+        config = UnhosedConfigParser()
+        config.read_string(text)
+        instance = self._makeOne()
+        self.assertRaises(ValueError,instance.process_groups_from_parser,config)
+
     def test_heterogeneous_process_groups_from_parser(self):
         text = lstrip("""\
         [program:one]
@@ -651,6 +714,37 @@ class TestProcessConfig(unittest.TestCase):
         recorder = instance.make_stdout_recorder()
         from supervisor.recorders import LoggingRecorder
         self.assertEqual(recorder.capturelog, None)
+
+class ProcessGroupConfigTests(unittest.TestCase):
+    def _getTargetClass(self):
+        from supervisor.options import ProcessGroupConfig
+        return ProcessGroupConfig
+
+    def _makeOne(self, options, name, priority, pconfigs):
+        return self._getTargetClass()(options, name, priority, pconfigs)
+
+    def test_ctor(self):
+        options = DummyOptions()
+        instance = self._makeOne(options, 'whatever', 999, [])
+        self.assertEqual(instance.options, options)
+        self.assertEqual(instance.name, 'whatever')
+        self.assertEqual(instance.priority, 999)
+        self.assertEqual(instance.process_configs, [])
+    
+    def test_after_setuid(self):
+        options = DummyOptions()
+        pconfigs = [DummyPConfig(options, 'process1', '/bin/process1')]
+        instance = self._makeOne(options, 'whatever', 999, pconfigs)
+        instance.after_setuid()
+        self.assertEqual(pconfigs[0].autochildlogs_created, True)
+
+    def test_make_group(self):
+        options = DummyOptions()
+        pconfigs = [DummyPConfig(options, 'process1', '/bin/process1')]
+        instance = self._makeOne(options, 'whatever', 999, pconfigs)
+        group = instance.make_group()
+        from supervisor.process import ProcessGroup
+        self.assertEqual(group.__class__, ProcessGroup)
             
 class BasicAuthTransportTests(unittest.TestCase):
     def _getTargetClass(self):