v2.5.0

Author: acn-ericlaw

CHANGELOG.md

@@ -5,6 +5,24 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+---
+## Version 2.5.0, 9/24/2022
+
+### Added
+
+N/A
+
+### Removed
+
+N/A
+
+### Changed
+
+For consistency,
+
+1. Refactor post office APIs for sending single event request as "send_request".
+2. Update platform APIs for sending event request and parallel requests as "send_request" and "send_parallel_requests"
+
 ---
 ## Version 2.3.6, 6/16/2022
 

README.md

@@ -147,7 +147,7 @@ try:
     if isinstance(result, EventEnvelope):
         print('Received RPC response:')
         print("HEADERS =", result.get_headers(), ", BODY =", result.get_body(),
-              ", STATUS =",  result.get_status(),
+              ", STATUS =", result.get_status(),
               ", EXEC =", result.get_exec_time(), ", ROUND TRIP =", result.get_round_trip(), "ms")
 except TimeoutError as e:
     print("Exception: ", str(e))

docs/guides/CHAPTER-1.md

@@ -80,19 +80,19 @@ try:
     if isinstance(result, EventEnvelope):
         print('Received RPC response:')
         print("HEADERS =", result.get_headers(), ", BODY =", result.get_body(),
-                ", STATUS =",  result.get_status(),
-                ", EXEC =", result.get_exec_time(), ", ROUND TRIP =", result.get_round_trip(), "ms")
+              ", STATUS =", result.get_status(),
+              ", EXEC =", result.get_exec_time(), ", ROUND TRIP =", result.get_round_trip(), "ms")
 except TimeoutError as e:
     print("Exception: ", str(e))
 
 # for async call
 po.send('hello.world.1', headers={'one': 1}, body='hello world one')
 ```
 
-## Massive parallel processing
+## Parallel processing
 
 A function is invoked when an event happens. Before the event arrives, the function is just an entry in a routing 
-table and it does not consume any additional resources like threads.
+table, and it does not consume any additional resources like threads.
 
 All functions are running in parallel without special coding. Behind the curtain, the system uses Python futures and 
 asyncio event loops for very efficient function execution.

docs/guides/CHAPTER-3.md

@@ -26,17 +26,23 @@ The Mercury framework is 100% event-driven and all communications are asynchrono
 it suspends the calling function and uses temporary Inbox as a callback function. The called function will send 
 the reply to the callback function which in turns wakes up the calling function.
 
-To make a RPC call, you can use the `request` method.
+To make a RPC call, you can use the `request` or the `send_request` method. 
+With the latter, you can send the request as an event and set event metadata such as correlation-ID.
 
 ```python
-request(self, route: str, timeout_seconds: float,
-                headers: dict = None, body: any = None,
-                correlation_id: str = None) -> EventEnvelope
+def request(self, route: str, timeout_seconds: float,
+              headers: dict = None, body: any = None,
+              correlation_id: str = None) -> EventEnvelope:
 
 # example
-result = po.request('hello.world.2', 2.0, headers={'some_key': 'some_value'}, body='hello world')
+result = po.send_request('hello.world.2', 2.0, headers={'some_key': 'some_value'}, body='hello world')
 print(result.get_body())
 
+# send request using an event directly
+def send_request(self, event: EventEnvelope, timeout_seconds: float) -> EventEnvelope:
+
+# example
+result = po.send_request(event, 2.0)
 ```
 
 Note that Mercury supports Python primitive or dictionary in the message body. If you put other object, it may throw 
@@ -47,7 +53,7 @@ serialization exception or the object may become empty.
 To make an asynchronous call, use the `send` method.
 
 ```python
-send(self, route: str, headers: dict = None, body: any = None, reply_to: str = None, me=True) -> None
+def send(self, route: str, headers: dict = None, body: any = None, reply_to: str = None, me=True) -> None""
 ```
 
 You may put key-value pairs in the "headers" field for holding parameters. For message payload, put Python primitive 
@@ -56,7 +62,7 @@ or dictionary in the "body" field.
 ### Deferred delivery
 
 ```python
-send_later(self, route: str, headers: dict = None, body: any = None, seconds: float = 1.0) -> None
+def send_later(self, route: str, headers: dict = None, body: any = None, seconds: float = 1.0) -> None:
 ```
 
 ### Call-back
@@ -148,7 +154,7 @@ Broadcast is the easiest way to do "pub/sub". To broadcast an event to multiple
 use the `broadcast` method.
 
 ```python
-broadcast(self, route: str, headers: dict = None, body: any = None) -> None
+def broadcast(self, route: str, headers: dict = None, body: any = None) -> None:
 
 # example
 po.broadcast("hello.world.1", body="this is a broadcast message from "+platform.get_origin())
@@ -160,7 +166,7 @@ po.broadcast("hello.world.1", body="this is a broadcast message from "+platform.
 You can perform join-n-fork RPC calls using a parallel version of the request, `parallel_request` method.
 
 ```python
-parallel_request(self, events: list, timeout_seconds: float) -> list
+def parallel_request(self, events: list, timeout_seconds: float) -> list:
 
 # illustrate parallel RPC requests
 event_list = list()
@@ -172,8 +178,8 @@ try:
         print('Received', len(result), 'RPC responses:')
         for res in result:
             print("HEADERS =", res.get_headers(), ", BODY =", res.get_body(),
-                    ", STATUS =",  res.get_status(),
-                    ", EXEC =", res.get_exec_time(), ", ROUND TRIP =", res.get_round_trip(), "ms")
+                  ", STATUS =", res.get_status(),
+                  ", EXEC =", res.get_exec_time(), ", ROUND TRIP =", res.get_round_trip(), "ms")
 except TimeoutError as e:
     print("Exception: ", str(e))
 ```
@@ -188,14 +194,14 @@ However, if you want to do store-n-forward pub/sub for certain use cases, you ma
 Following are some useful pub/sub API:
 
 ```python
-def feature_enabled()
-def create_topic(topic: str)
-def delete_topic(topic: str)
-def publish(topic: str, headers: dict = None, body: any = None)
-def subscribe(self, topic: str, route: str, parameters: list = None)
-def unsubscribe(self, topic: str, route: str)
-def exists(topic: str)
-def list_topics()
+def feature_enabled():
+def create_topic(topic: str):
+def delete_topic(topic: str):
+def publish(topic: str, headers: dict = None, body: any = None):
+def subscribe(self, topic: str, route: str, parameters: list = None):
+def unsubscribe(self, topic: str, route: str):
+def exists(topic: str):
+def list_topics():
 
 ```
 Some pub/sub engine would require additional parameters when subscribing a topic. For Kafka, you must provide the 
@@ -222,7 +228,7 @@ serialization yourself. The payload can be a dict, bool, str, int or float.
 To check if a target service is available, you can use the `exists` method.
 
 ```python
-exists(self, routes: any)
+def exists(self, routes: any) -> bool:
 
 # input can be a route name or a list of routes
 # it will return true only when all routes are available

examples/standalone-tracing.py

@@ -29,7 +29,6 @@
 def tracing(headers: dict, body: any):
     # no instance parameter because this is a singleton
     log.info(f'TRACE {headers}')
-    return body
 
 
 def hello(headers: dict, body: any, instance: int):
@@ -54,7 +53,7 @@ def main():
         trace_path = 'GET /api/hello/world'
         event = EventEnvelope().set_to("hello.world").set_header('some_key', 'some_value').set_body('hello world')
         event.set_trace(trace_id, trace_path).set_from('this.demo')
-        result = po.single_request(event, 2.0)
+        result = po.send_request(event, 2.0)
         if isinstance(result, EventEnvelope):
             log.info('Received RPC response:')
             log.info(f'HEADERS = {result.get_headers()}, BODY = {result.get_body()}, STATUS = {result.get_status()}, '

examples/stream-demo.py

@@ -65,9 +65,9 @@ def main():
 
     input_stream.close()
     #
-    # This will keep the main thread running in the background.
-    # We can use Control-C or KILL signal to stop the application.
-    platform.run_forever()
+    # Stop platform after demo
+    #
+    platform.stop()
 
 
 if __name__ == '__main__':

mercury/platform.py

@@ -600,7 +600,7 @@ def route_instances(self, route: str) -> int:
         else:
             return 0
 
-    def parallel_request(self, events: list, timeout_seconds: float):
+    def send_parallel_requests(self, events: list, timeout_seconds: float):
         timeout_value = self.util.get_float(timeout_seconds)
         if timeout_value <= 0:
             raise ValueError('timeout value in seconds must be positive number')
@@ -610,7 +610,7 @@ def parallel_request(self, events: list, timeout_seconds: float):
             raise ValueError('event list is empty')
         if len(events) == 1:
             result = list()
-            result.append(self.request(events[0], timeout_value))
+            result.append(self.send_request(events[0], timeout_value))
             return result
         for evt in events:
             if not isinstance(evt, EventEnvelope):
@@ -650,12 +650,12 @@ def parallel_request(self, events: list, timeout_seconds: float):
                     if len(result_list) == len(events):
                         return result_list
                 except Empty:
-                    raise TimeoutError(f'Requests timeout for {round(timeout_value, 3)} seconds. '
+                    raise TimeoutError(f'Request timeout for {round(timeout_value, 3)} seconds. '
                                        f'Expect: {total_requests} responses, actual: {len(result_list)}')
         finally:
             inbox.close()
 
-    def request(self, event: EventEnvelope, timeout_seconds: float):
+    def send_request(self, event: EventEnvelope, timeout_seconds: float):
         timeout_value = self.util.get_float(timeout_seconds)
         if timeout_value <= 0:
             raise ValueError('timeout value in seconds must be positive number')
@@ -724,15 +724,15 @@ def send_event(self, event: EventEnvelope, broadcast=False) -> None:
     def send_event_later(self, event: EventEnvelope, delay_in_seconds: float) -> None:
         self._loop.call_later(delay_in_seconds, self.send_event, event)
 
-    def exists(self, routes: any):
+    def exists(self, routes: any) -> bool:
         if isinstance(routes, str):
             single_route = routes
             if self.has_route(single_route):
                 return True
             if self.cloud_ready():
                 event = EventEnvelope()
                 event.set_to(self.SERVICE_QUERY).set_header('type', 'find').set_header('route', single_route)
-                result = self.request(event, 8.0)
+                result = self.send_request(event, 8.0)
                 if isinstance(result, EventEnvelope):
                     if result.get_body() is not None:
                         return result.get_body()
@@ -749,7 +749,7 @@ def exists(self, routes: any):
                     event = EventEnvelope()
                     event.set_to(self.SERVICE_QUERY).set_header('type', 'find')
                     event.set_header('route', '*').set_body(routes)
-                    result = self.request(event, 8.0)
+                    result = self.send_request(event, 8.0)
                     if isinstance(result, EventEnvelope) and result.get_body() is not None:
                         return result.get_body()
         return False

mercury/system/object_stream.py

@@ -21,7 +21,6 @@
 
 
 class ObjectStreamIO:
-
     STREAM_IO_MANAGER = 'object.streams.io'
 
     def __init__(self, expiry_seconds: int = 1800):

mercury/system/po.py

@@ -141,16 +141,16 @@ def request(self, route: str, timeout_seconds: float,
             event.set_body(body)
         if correlation_id is not None:
             event.set_correlation_id(str(correlation_id))
-        response = self.platform.request(event, timeout_seconds)
+        response = self.platform.send_request(event, timeout_seconds)
         if isinstance(response, EventEnvelope):
             if response.get_tag('exception') is None:
                 return response
             else:
                 raise AppException(response.get_status(), response.get_body())
         raise ValueError(f'Expect response is EventEnvelope, actual: ({response})')
 
-    def single_request(self, event: EventEnvelope, timeout_seconds: float):
-        response = self.platform.request(event, timeout_seconds)
+    def send_request(self, event: EventEnvelope, timeout_seconds: float) -> EventEnvelope:
+        response = self.platform.send_request(event, timeout_seconds)
         if isinstance(response, EventEnvelope):
             if response.get_tag('exception') is None:
                 return response
@@ -159,7 +159,7 @@ def single_request(self, event: EventEnvelope, timeout_seconds: float):
         raise ValueError(f'Expect response is EventEnvelope, actual: ({response})')
 
     def parallel_request(self, events: list, timeout_seconds: float) -> list:
-        return self.platform.parallel_request(events, timeout_seconds)
+        return self.platform.send_parallel_requests(events, timeout_seconds)
 
-    def exists(self, routes: any):
+    def exists(self, routes: any) -> bool:
         return self.platform.exists(routes)

setup.py

@@ -4,7 +4,7 @@
 from setuptools import setup
 
 setup(name='mercury',
-      version='2.3.6',
+      version='2.5.0',
       description='Python Language pack for Mercury',
       author='Eric Law',
       author_email='[email protected]',
@@ -16,13 +16,10 @@
       package_dir={'mercury': 'mercury'},
       package_data={'mercury': ['resources/application.yml']},
       license='Apache 2.0',
-      python_requires='>=3.6.7',
+      python_requires='>=3.8.0',
       install_requires=['aiohttp', 'msgpack-python', 'PyYAML', 'pytest'],
       classifiers=[
           'Programming Language :: Python :: 3',
-          'Programming Language :: Python :: 3.5',
-          'Programming Language :: Python :: 3.6',
-          'Programming Language :: Python :: 3.7',
           'Programming Language :: Python :: 3.8',
           'Programming Language :: Python :: 3.9',
           'Programming Language :: Python :: 3.10',