[EventDispatcher] Guard against broken pipe and eof (#1074)

* Guard against broken pipe and eof

* refactor

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Author: abhinavsingh

proxy/core/event/dispatcher.py

@@ -52,22 +52,26 @@ def __init__(
 
     def handle_event(self, ev: Dict[str, Any]) -> None:
         if ev['event_name'] == eventNames.SUBSCRIBE:
-            self.subscribers[ev['event_payload']['sub_id']] = \
-                ev['event_payload']['conn']
+            sub_id = ev['event_payload']['sub_id']
+            self.subscribers[sub_id] = ev['event_payload']['conn']
             # send ack
-            ev['event_payload']['conn'].send({
-                'event_name': eventNames.SUBSCRIBED,
-            })
+            if not self._send(
+                sub_id, {
+                    'event_name': eventNames.SUBSCRIBED,
+                },
+            ):
+                self._close_and_delete(sub_id)
         elif ev['event_name'] == eventNames.UNSUBSCRIBE:
-            if ev['event_payload']['sub_id'] in self.subscribers:
+            sub_id = ev['event_payload']['sub_id']
+            if sub_id in self.subscribers:
                 # send ack
                 logger.debug('unsubscription request ack sent')
-                self.subscribers[ev['event_payload']['sub_id']].send({
-                    'event_name': eventNames.UNSUBSCRIBED,
-                })
-                # close conn and delete subscriber
-                self.subscribers[ev['event_payload']['sub_id']].close()
-                del self.subscribers[ev['event_payload']['sub_id']]
+                self._send(
+                    sub_id, {
+                        'event_name': eventNames.UNSUBSCRIBED,
+                    },
+                )
+                self._close_and_delete(sub_id)
             else:
                 logger.info(
                     'unsubscription request ack not sent, subscriber already gone',
@@ -87,11 +91,12 @@ def run(self) -> None:
                     self.run_once()
                 except queue.Empty:
                     pass
-        except (BrokenPipeError, EOFError, KeyboardInterrupt):
+        except KeyboardInterrupt:
             pass
         except Exception as e:
             logger.exception('Dispatcher exception', exc_info=e)
         finally:
+            logger.info('Dispatcher shutdown')
             # Send shutdown message to all active subscribers
             self._broadcast({
                 'event_name': eventNames.DISPATCHER_SHUTDOWN,
@@ -106,7 +111,26 @@ def _broadcast(self, ev: Dict[str, Any]) -> None:
                 logger.warning(
                     'Subscriber#%s broken pipe', sub_id,
                 )
-                self.subscribers[sub_id].close()
+                self._close(sub_id)
                 broken_pipes.append(sub_id)
         for sub_id in broken_pipes:
             del self.subscribers[sub_id]
+
+    def _close_and_delete(self, sub_id: str) -> None:
+        self._close(sub_id)
+        del self.subscribers[sub_id]
+
+    def _close(self, sub_id: str) -> None:
+        try:
+            self.subscribers[sub_id].close()
+        except Exception:   # noqa: S110
+            pass
+
+    def _send(self, sub_id: str, payload: Any) -> bool:
+        done = False
+        try:
+            self.subscribers[sub_id].send(payload)
+            done = True
+        except (BrokenPipeError, EOFError):
+            pass
+        return done