[Examples] Fix broken examples (#1077)

* Fix broken examples

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

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

* Fix mypy

* Pylint ignore

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

Author: abhinavsingh

examples/pubsub_eventing.py

@@ -66,42 +66,37 @@ def publisher_process(
         # Create a subscriber.
         # Internally, subscribe will start a separate thread
         # to receive incoming published messages.
-        subscriber = EventSubscriber(event_manager.queue, callback=on_event)
-        subscriber.setup()
-
-        # Start a publisher process to demonstrate safe exchange
-        # of messages between processes.
-        publisher_shutdown_event = multiprocessing.Event()
-        publisher = multiprocessing.Process(
-            target=publisher_process, args=(
-                publisher_shutdown_event, event_manager.queue, ),
-        )
-        publisher.start()
-
-        # Dispatch event from main process too
-        # to demonstrate safe exchange of messages
-        # between threads.
-        try:
-            while True:
-                event_manager.queue.publish(
-                    request_id='1234',
-                    event_name=eventNames.WORK_STARTED,
-                    event_payload={'time': time.time()},
-                    publisher_id='eventing_pubsub_main',
-                )
-        except KeyboardInterrupt:
-            logger.info('bye!!!')
-        finally:
-            # Stop publisher process
-            publisher_shutdown_event.set()
-            publisher.join()
-            # Stop subscriber thread
-            subscriber.unsubscribe()
-            logger.info(
-                'Received {0} events from main thread, {1} events from another process, in {2} seconds'.format(
-                    num_events_received[0], num_events_received[1], time.time(
-                    ) - start_time,
-                ),
+        with EventSubscriber(event_manager.queue, callback=on_event) as subscriber:
+            # Start a publisher process to demonstrate safe exchange
+            # of messages between processes.
+            publisher_shutdown_event = multiprocessing.Event()
+            publisher = multiprocessing.Process(
+                target=publisher_process, args=(
+                    publisher_shutdown_event, event_manager.queue, ),
             )
-    if subscriber:
-        subscriber.shutdown(do_unsubscribe=False)
+            publisher.start()
+
+            # Dispatch event from main process too
+            # to demonstrate safe exchange of messages
+            # between threads.
+            try:
+                while True:
+                    event_manager.queue.publish(
+                        request_id='1234',
+                        event_name=eventNames.WORK_STARTED,
+                        event_payload={'time': time.time()},
+                        publisher_id='eventing_pubsub_main',
+                    )
+            except KeyboardInterrupt:
+                logger.info('KBE!!!')
+            finally:
+                # Stop publisher process
+                publisher_shutdown_event.set()
+                publisher.join()
+                logger.info(
+                    'Received {0} events from main thread, {1} events from another process, in {2} seconds'.format(
+                        num_events_received[0], num_events_received[1], time.time(
+                        ) - start_time,
+                    ),
+                )
+    logger.info('Done!!!')

examples/ssl_echo_client.py

@@ -8,21 +8,29 @@
     :copyright: (c) 2013-present by Abhinav Singh and contributors.
     :license: BSD, see LICENSE for more details.
 """
+import ssl
 import logging
 
 from proxy.core.connection import TcpServerConnection
-from proxy.common.constants import DEFAULT_BUFFER_SIZE
+from proxy.common.constants import DEFAULT_LOG_FORMAT, DEFAULT_BUFFER_SIZE
 
 
+logging.basicConfig(level=logging.INFO, format=DEFAULT_LOG_FORMAT)
+
 logger = logging.getLogger(__name__)
 
 if __name__ == '__main__':
-    client = TcpServerConnection('::', 12345)
+    client = TcpServerConnection('127.0.0.1', 12345)
     client.connect()
-    client.wrap('example.com', ca_file='ca-cert.pem')
-    # wrap() will by default set connection to nonblocking
-    # flip it back to blocking
-    client.connection.setblocking(True)
+    client.wrap(
+        None,  # 'localhost',
+        ca_file='ca-cert.pem',
+        # For self-signed certs you will have
+        # to disable verification.  Or you can
+        # add your CA certificate in the CA bundle
+        # and then enable verify.
+        verify_mode=ssl.VerifyMode.CERT_NONE,
+    )
     try:
         while True:
             client.send(b'hello')

examples/ssl_echo_server.py

@@ -52,8 +52,8 @@ def main() -> None:
         threadless=True,
         num_workers=1,
         port=12345,
-        keyfile='https-key.pem',
-        certfile='https-signed-cert.pem',
+        key_file='https-key.pem',
+        cert_file='https-signed-cert.pem',
     ):
         try:
             while True:

examples/tcp_echo_client.py

@@ -11,13 +11,15 @@
 import logging
 
 from proxy.common.utils import socket_connection
-from proxy.common.constants import DEFAULT_BUFFER_SIZE
+from proxy.common.constants import DEFAULT_LOG_FORMAT, DEFAULT_BUFFER_SIZE
 
 
+logging.basicConfig(level=logging.INFO, format=DEFAULT_LOG_FORMAT)
+
 logger = logging.getLogger(__name__)
 
 if __name__ == '__main__':
-    with socket_connection(('::', 12345)) as client:
+    with socket_connection(('127.0.0.1', 12345)) as client:
         while True:
             client.send(b'hello')
             data = client.recv(DEFAULT_BUFFER_SIZE)

examples/websocket_client.py

@@ -14,8 +14,11 @@
 from proxy.http.websocket import (
     WebsocketFrame, WebsocketClient, websocketOpcodes,
 )
+from proxy.common.constants import DEFAULT_LOG_FORMAT
 
 
+logging.basicConfig(level=logging.INFO, format=DEFAULT_LOG_FORMAT)
+
 # globals
 client: WebsocketClient
 last_dispatch_time: float
@@ -47,9 +50,9 @@ def on_message(frame: WebsocketFrame) -> None:
 if __name__ == '__main__':
     # Constructor establishes socket connection
     client = WebsocketClient(
-        b'echo.websocket.org',
-        80,
-        b'/',
+        b'localhost',
+        8899,
+        b'/ws-route-example',
         on_message=on_message,
     )
     # Perform handshake

proxy/common/utils.py

@@ -213,12 +213,15 @@ def find_http_line(raw: bytes) -> Tuple[Optional[bytes], bytes]:
 
 
 def wrap_socket(
-    conn: socket.socket, keyfile: str,
-    certfile: str,
+        conn: socket.socket,
+        keyfile: str,
+        certfile: str,
+        cafile: Optional[str] = None,
 ) -> ssl.SSLSocket:
     """Use this to upgrade server_side socket to TLS."""
     ctx = ssl.create_default_context(
         ssl.Purpose.CLIENT_AUTH,
+        cafile=cafile,
     )
     ctx.options |= ssl.OP_NO_SSLv2 | ssl.OP_NO_SSLv3 | ssl.OP_NO_TLSv1 | ssl.OP_NO_TLSv1_1
     ctx.verify_mode = ssl.CERT_NONE

proxy/core/base/tcp_server.py

@@ -185,8 +185,19 @@ async def handle_readables(self, readables: Readables) -> bool:
         if self.work.connection.fileno() in readables:
             try:
                 data = self.work.recv(self.flags.client_recvbuf_size)
+            except ConnectionResetError:
+                logger.info(
+                    'Connection reset by client {0}'.format(
+                        self.work.address,
+                    ),
+                )
+                return True
             except TimeoutError:
-                logger.info('Client recv timeout error')
+                logger.info(
+                    'Client recv timeout error {0}'.format(
+                        self.work.address,
+                    ),
+                )
                 return True
             if data is None:
                 logger.debug(

proxy/core/connection/server.py

@@ -45,15 +45,19 @@ def connect(
 
     def wrap(
             self,
-            hostname: str,
+            hostname: Optional[str] = None,
             ca_file: Optional[str] = None,
             as_non_blocking: bool = False,
+            # Ref https://github.com/PyCQA/pylint/issues/3691
+            verify_mode: ssl.VerifyMode = ssl.VerifyMode.CERT_REQUIRED,     # pylint: disable=E1101
     ) -> None:
         ctx = ssl.create_default_context(
-            ssl.Purpose.SERVER_AUTH, cafile=ca_file,
+            ssl.Purpose.SERVER_AUTH,
+            cafile=ca_file,
         )
         ctx.options |= ssl.OP_NO_SSLv2 | ssl.OP_NO_SSLv3 | ssl.OP_NO_TLSv1 | ssl.OP_NO_TLSv1_1
-        ctx.check_hostname = True
+        ctx.check_hostname = hostname is not None
+        ctx.verify_mode = verify_mode
         self.connection.setblocking(True)
         self._conn = ctx.wrap_socket(
             self.connection,

proxy/core/event/dispatcher.py

@@ -99,11 +99,11 @@ def run(self) -> None:
         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,
             })
+            logger.info('Dispatcher shutdown')
 
     def _broadcast(self, ev: Dict[str, Any]) -> None:
         broken_pipes: List[str] = []

proxy/core/event/queue.py

@@ -51,7 +51,7 @@ def publish(
         event_payload: Dict[str, Any],
         publisher_id: Optional[str] = None,
     ) -> None:
-        self.queue.put_nowait({
+        self.queue.put({
             'process_id': os.getpid(),
             'thread_id': threading.get_ident(),
             'event_timestamp': time.time(),
@@ -71,7 +71,7 @@ def subscribe(
         sub_id is a subscription identifier which must be globally
         unique.  channel MUST be a multiprocessing connection.
         """
-        self.queue.put_nowait({
+        self.queue.put({
             'event_name': eventNames.SUBSCRIBE,
             'event_payload': {'sub_id': sub_id, 'conn': channel},
         })
@@ -81,7 +81,7 @@ def unsubscribe(
             sub_id: str,
     ) -> None:
         """Unsubscribe by subscriber id."""
-        self.queue.put_nowait({
+        self.queue.put({
             'event_name': eventNames.UNSUBSCRIBE,
             'event_payload': {'sub_id': sub_id},
         })