Release v0.13.0

.pre-commit-config.yaml

@@ -6,7 +6,7 @@ repos:
         args: [--safe, --quiet]
         language_version: python3
 -   repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v2.1.0
+    rev: v2.4.0
     hooks:
     -   id: trailing-whitespace
     -   id: end-of-file-fixer

.travis.yml

@@ -5,8 +5,8 @@ services:
 
 language: python
 python:
-  - "3.6"
   - "3.7"
+  - "3.8"
 
 install:
   - pip install -r requirements/local.txt

README.md

@@ -48,14 +48,13 @@ a Redis Pub/Sub channel for immediate delivery.
 
     ```python
     from wtfix.apps.base import MessageTypeHandlerApp, on
-    from wtfix.protocol.common import MsgType
     from wtfix.conf import settings
 
     logger = settings.logger
 
     class SecretAlgoTradingRecipe(MessageTypeHandlerApp):
 
-        @on(MsgType.Logon)  # Only invoked when 'Logon (type A)' messages are received.
+        @on(settings.protocol.MsgType.Logon)  # Only invoked when 'Logon (type A)' messages are received.
         def on_logon(self, message):
             self.send_security_definition_request()
             return message
@@ -72,7 +71,6 @@ used message attributes.
 
     ```python
     >>> from wtfix.message import admin
-    >>> from wtfix.protocol.common import Tag
 
     # Instantiate a new 'Logon' message
     >>> logon_msg = admin.LogonMessage("my_username", "my_password", heartbeat_int=30)
@@ -102,7 +100,7 @@ used message attributes.
     >>> logon_msg[108]  # Using old school tag number
     Field(108, '30')
 
-    >>> logon_msg[Tag.HeartBtInt]  # Using the tag name as per the FIX specification
+    >>> logon_msg[settings.protocol.Tag.HeartBtInt]  # Using the tag name as per the FIX specification
     Field(108, '30')
 
     >>> logon_msg.HeartBtInt  # Using tag name shortcut

config/settings/base.py

@@ -28,10 +28,6 @@
 # ------------------------------------------------------------------------------
 REDIS_URI = os.getenv("REDIS_URI", "redis://localhost:6379/0")
 
-# MESSAGE STORE
-# ------------------------------------------------------------------------------
-MESSAGE_STORE = os.getenv("MESSAGE_STORE", "wtfix.apps.store.MemoryStore")
-
 # CONNECTIONS
 # ------------------------------------------------------------------------------
 CONNECTIONS = {
@@ -45,6 +41,7 @@
         "PASSWORD": os.getenv("PASSWORD"),
         # APPS
         "PIPELINE_APPS": [
+            "wtfix.apps.utils.PipelineTerminationApp",
             "wtfix.apps.api.rest.RESTfulServiceApp",
             "wtfix.apps.brokers.RedisPubSubApp",
             "wtfix.apps.admin.HeartbeatApp",
@@ -57,6 +54,14 @@
             "wtfix.apps.wire.WireCommsApp",
             "wtfix.apps.sessions.ClientSessionApp",
         ],
+        # PROTOCOL
+        "PROTOCOL": "wtfix.protocol.fix._44.spec.FIX44Protocol",
+        # MESSAGE STORE
+        "MESSAGE_STORE": {
+            "CLASS": "wtfix.apps.store.MemoryStore",
+            "ENCODER": "wtfix.core.encoders.JSONMessageEncoder",
+            "DECODER": "wtfix.core.decoders.JSONMessageDecoder",
+        },
         # REPEATING GROUPS
         "GROUP_TEMPLATES": {},
     }

config/settings/test.py

@@ -1,4 +1,4 @@
-from wtfix.protocol.common import Tag
+from wtfix.core.klass import get_class_from_module_string
 from .local import *  # noqa
 
 # GENERAL
@@ -17,7 +17,9 @@
 CONNECTIONS["default"]["PASSWORD"] = "TEST_PASSWORD"
 
 # REPEATING GROUPS
+# Get protocol Type so that we can configure repeating groups  on a per-protocol basis
+protocol = get_class_from_module_string(CONNECTIONS["default"]["PROTOCOL"])
 CONNECTIONS["default"]["GROUP_TEMPLATES"] = {
     # Routing IDs
-    Tag.NoRoutingIDs: {"*": [Tag.RoutingType, Tag.RoutingID]}
+    protocol.Tag.NoRoutingIDs: {"*": [protocol.Tag.RoutingType, protocol.Tag.RoutingID]}
 }

config/wsgi.py

@@ -20,10 +20,6 @@
 
 from flask import Flask
 
-from unsync import (
-    unsync,
-)  # Import unsync to set event loop and start ambient unsync thread
-
 from wtfix.apps.api.rest import RESTfulServiceApp
 from wtfix.conf import settings
 from wtfix.pipeline import BasePipeline

docs/changelog.md

@@ -2,6 +2,26 @@
 
 This changelog is used to track all major changes to WTFIX.
 
+## v0.13.0 (2019-11-14)
+
+**Enhancements**
+
+- BREAKING CONFIG CHANGES (please update your settings files in `wtfix.config.settings`):
+    - The MESSAGE_STORE parameters now form part of the CONNECTION section. This allows individual message stores to
+      be configured when multiple connections need to be run simultaneously.
+    - Add configuration option for specifying which JSON encoder / decoder to use when adding messages to a message
+      store.
+    - The FIX protocol and version can now be configured for individual CONNECTIONS. This lays the foundation for
+      supporting various different protocols and versions in the future. FIX 4.4 is currently the default.
+
+- Add `PipelineTerminationApp` and use `del` to encourage the Python interpreter to garbage collect a message once it
+  has reached either end of the pipeline.
+- Upgrade aioredis dependency to version 1.3
+- Remove dependency on 'unsync' which has become largely redundant with the new async features released as part
+  of Python 3.7.
+- Now requires Python >= 3.7.
+
+
 ## v0.12.4 (2019-09-02)
 
 **Fixes**

requirements/base.txt

@@ -1,6 +1,5 @@
 python-dotenv~=0.10.3  # https://github.com/theskumar/python-dotenv
-unsync~=1.2  # https://github.com/alex-sherman/unsync
 flask-restful~=0.3.7  # https://github.com/flask-restful/flask-restful
 requests~=2.22.0  # https://github.com/kennethreitz/requests
 gunicorn~=19.9.0  # https://gunicorn.org
-aioredis~=1.2  # https://github.com/aio-libs/aioredis
+aioredis~=1.3  # https://github.com/aio-libs/aioredis

run_client.py

@@ -16,16 +16,13 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import argparse
-import functools
+import asyncio
 import logging
 import os
 import sys
 import signal
 from asyncio import futures
 
-# Import unsync to set event loop and start ambient unsync thread
-from unsync import unsync  # noqa
-
 from wtfix.conf import settings
 from wtfix.core.exceptions import ImproperlyConfigured
 from wtfix.pipeline import BasePipeline
@@ -35,14 +32,15 @@
 parser = argparse.ArgumentParser(description="Start a FIX connection")
 
 try:
-    default_connection = settings.default_connection_name
+    # If only one connection has been configured then we have a safe default to fall back to.
+    default_connection_name = settings.default_connection_name
 except ImproperlyConfigured:
-    default_connection = None
+    default_connection_name = None
 
 parser.add_argument(
     "--connection",
-    default=default_connection,
-    help=f"the configuration settings to use for the connection (default: '{default_connection}')",
+    default=default_connection_name,
+    help=f"the configuration settings to use for the connection (default: '{default_connection_name}')",
 )
 
 parser.add_argument(
@@ -52,41 +50,42 @@
 )
 
 
-@unsync
 async def graceful_shutdown(sig_name_, pipeline):
     logger.info(f"Received signal {sig_name_}! Initiating shutdown...")
 
     try:
-        task = pipeline.stop()
-        await task
-        task.cancel()  # Terminate the task once we're done
+        await pipeline.stop()
 
     except futures.CancelledError as e:
         logger.error(f"Cancelled: connection terminated abnormally! ({e})")
         sys.exit(os.EX_UNAVAILABLE)
 
 
-if __name__ == "__main__":
+async def main():
     logging.basicConfig(
         level=settings.LOGGING_LEVEL,
         format="%(asctime)s - %(threadName)s - %(module)s - %(levelname)s - %(message)s",
     )
 
     args = parser.parse_args()
+
     fix_pipeline = BasePipeline(
         connection_name=args.connection, new_session=args.new_session
     )
 
     try:
         # Graceful shutdown on termination signals.
-        # See: https://docs.python.org/3.6/library/asyncio-eventloop.html#set-signal-handlers-for-sigint-and-sigterm
+        # See: https://docs.python.org/3.7/library/asyncio-eventloop.html#set-signal-handlers-for-sigint-and-sigterm
+        loop = asyncio.get_running_loop()
         for sig_name in {"SIGINT", "SIGTERM"}:
-            unsync.loop.add_signal_handler(
+            loop.add_signal_handler(
                 getattr(signal, sig_name),
-                functools.partial(graceful_shutdown, sig_name, fix_pipeline),
+                lambda: asyncio.ensure_future(
+                    graceful_shutdown(sig_name, fix_pipeline)
+                ),
             )
 
-        fix_pipeline.start().result()
+        await fix_pipeline.start()
 
     except futures.TimeoutError as e:
         logger.error(e)
@@ -110,7 +109,11 @@ async def graceful_shutdown(sig_name_, pipeline):
 
     finally:
         try:
-            fix_pipeline.stop().result()
+            await fix_pipeline.stop()
         except futures.CancelledError as e:
             logger.error(f"Cancelled: connection terminated abnormally! ({e})")
             sys.exit(os.EX_UNAVAILABLE)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

setup.py

@@ -9,7 +9,7 @@
 
 setup(
     name="wtfix",
-    version="0.12.4",
+    version="0.13.0",
     author="John Cass",
     author_email="john.cass77@gmail.com",
     description="The Pythonic Financial Information eXchange (FIX) client for humans.",
@@ -18,23 +18,22 @@
     url="https://github.com/jcass77/WTFIX",
     classifiers=[
         "Development Status :: 3 - Alpha",
-        "Programming Language :: Python :: 3.6",
         "Programming Language :: Python :: 3.7",
+        "Programming Language :: Python :: 3.8",
         "License :: OSI Approved :: GNU Lesser General Public License v3 (LGPLv3)",
         "Operating System :: OS Independent",
         "Intended Audience :: Financial and Insurance Industry",
     ],
     keywords="FIX financial information exchange",
     packages=find_packages(exclude=["contrib", "docs", "tests"]),
     install_requires=[
-        "python-dotenv~=0.10.1",
-        "unsync~=1.2",
+        "python-dotenv~=0.10.3",
         "flask-restful~=0.3.7",
-        "requests~=2.21.0",
+        "requests~=2.22.0",
         "gunicorn~=19.9.0",
-        "aioredis~=1.2",
+        "aioredis~=1.3",
     ],
-    python_requires=">=3.6",
+    python_requires=">=3.7",
     project_urls={
         "Bug Reports": "https://github.com/jcass77/WTFIX/issues",
         "Source": "https://github.com/jcass77/WTFIX/",