Repository: lwinterface/panini (language: python)

Issue:
Emulator

Default logger level 
By default it's warning
I would rather appreciate having INFO instead


Discussion / hints:

Fix patch:
diff --git a/.gitignore b/.gitignore
index b6f8e8a..0081739 100644
--- a/.gitignore
+++ b/.gitignore
@@ -132,4 +132,6 @@ dmypy.json
 
 **/logfiles
 **/logs
-**/log_config.json
\ No newline at end of file
+**/log_config.json
+**/test_logs
+**/test_logger_logs
\ No newline at end of file
diff --git a/docker-compose.yml b/docker-compose.yml
index 90fb2c2..5a2470a 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -4,6 +4,10 @@ networks:
   internet:
     name: "internet"
     driver: bridge
+  no-internet:
+    name: "no-internet"
+    internal: true
+    driver: bridge
 
 
 services:
@@ -15,8 +19,8 @@ services:
       - "4222"
       - "-m"
       - "8222"
-      - "-c"
-      - "/config/nats-server.conf"
+#      - "-c"
+#      - "/config/nats-server.conf"
     ports:
       - 4222:4222
       - 8222:8222
diff --git a/examples/emulator/listener_app.py b/examples/emulator/listener_app.py
new file mode 100644
index 0000000..f36f21f
--- /dev/null
+++ b/examples/emulator/listener_app.py
@@ -0,0 +1,36 @@
+from panini.app import App
+# from panini.emulator import WriterEmulatorMiddleware, ReaderEmulatorMiddleware
+from panini.emulator import WriterEmulatorMiddleware, ReaderEmulatorMiddleware
+from panini.nats_client.nats_client_interface import Msg
+
+app = App(
+    service_name="listener",
+    host="127.0.0.1",
+    port=4222
+)
+
+
+@app.listen("listener.store.listen")
+async def listen(message: Msg):
+    try:
+        print(message.subject, message.data)
+    except Exception as ex:
+        app.logger.exception(message.subject)
+
+
+@app.listen("listener.store.request")
+async def response(message: Msg):
+    try:
+        print(message.subject, message.data)
+        return {"data": "request"}
+    except Exception as ex:
+        app.logger.exception(message.subject)
+
+
+if __name__ == "__main__":
+    folder = "resources"
+    filename = "resources/events.listener.2021-03-22-16:30:18.jsonl"
+    # app.add_middleware(WriterEmulatorMiddleware, folder=folder)
+    app.add_middleware(ReaderEmulatorMiddleware, filename=filename)
+    # app.add_middleware(ReaderEmulatorMiddleware, prefix='prefix', run_emulator=False)
+    app.start()
diff --git a/examples/emulator/publisher_app.py b/examples/emulator/publisher_app.py
new file mode 100644
index 0000000..c9e0ac3
--- /dev/null
+++ b/examples/emulator/publisher_app.py
@@ -0,0 +1,44 @@
+import asyncio
+
+from panini.app import App
+from panini.emulator import WriterEmulatorMiddleware
+from panini.emulator import ReaderEmulatorMiddleware
+from panini.middleware.nats_timeout import NATSTimeoutMiddleware
+
+app = App(
+    service_name="publisher",
+    host="127.0.0.1",
+    port=4222
+)
+
+
+@app.task()
+async def request_task():
+    for i in range(10):
+        try:
+            print("request listener.store.request")
+
+            response = await app.request("listener.store.request", {"data": f"request.data.{i}"})
+            print('response', response)
+            await asyncio.sleep(1.5)
+        except Exception as ex:
+            app.logger.exception(ex)
+
+
+@app.task()
+async def publish_task():
+    for i in range(10):
+        try:
+            print("publish listener.store.listen")
+            await app.publish("listener.store.listen", {"data": f"publish.data.{i}"})
+            await asyncio.sleep(1)
+        except Exception as ex:
+            app.logger.exception(ex)
+
+
+if __name__ == "__main__":
+    # folder = "resources"
+    # filename = "events.publisher.2021-03-19-16:21:30.jsonl"
+    # app.add_middleware(ReaderEmulatorMiddleware, folder=folder, filename=f"{folder}/{filename}", compare_output=True)
+    # app.add_middleware(ReaderEmulatorMiddleware, prefix='prefix', run_emulator=False)
+    app.start()
diff --git a/examples/emulator/publisher_listener_app.py b/examples/emulator/publisher_listener_app.py
new file mode 100644
index 0000000..56b7e82
--- /dev/null
+++ b/examples/emulator/publisher_listener_app.py
@@ -0,0 +1,55 @@
+import asyncio
+
+from panini.app import App
+from panini.emulator import WriterEmulatorMiddleware
+
+app = App(
+    service_name="listener_publisher",
+    host="127.0.0.1",
+    port=4222
+)
+
+
+@app.listen("store.listen")
+async def listen(message):
+    try:
+        print(message.subject, message.data)
+    except Exception as ex:
+        app.logger.exception(message.subject)
+
+
+@app.listen("store.request")
+async def response(message):
+    try:
+        print(message.subject, message.data)
+        return {"data": "request"}
+    except Exception as ex:
+        app.logger.exception(message.subject)
+
+
+@app.task()
+async def request_task():
+    for i in range(10):
+        try:
+            response = await app.request("store.request", {"data": f"request.data.{i}"})
+            print('response', response)
+            await asyncio.sleep(1.5)
+        except Exception as ex:
+            app.logger.exception(ex)
+
+
+@app.task()
+async def publish_task():
+    for i in range(10):
+        try:
+            await app.publish("store.listen", {"data": f"publish.data.{i}"})
+            await asyncio.sleep(1)
+        except Exception as ex:
+            app.logger.exception(ex)
+
+
+if __name__ == "__main__":
+    folder = "resources"
+    filename = "resources/events.listener_publisher.2021-03-19-16:22:26.jsonl"
+    app.add_middleware(WriterEmulatorMiddleware, folder=folder)
+    app.start()
diff --git a/examples/emulator/reader.py b/examples/emulator/reader.py
new file mode 100644
index 0000000..2ca5a6e
--- /dev/null
+++ b/examples/emulator/reader.py
@@ -0,0 +1,46 @@
+import asyncio
+import time
+
+from panini import app
+from panini.emulator.emulator_client import EmulatorClient
+from panini.emulator.reader_emulator_middleware import ReaderEmulatorMiddleware
+from panini.utils.helper import start_process
+
+app = app.App(
+    service_name="reader",
+    host="127.0.0.1",
+    port=4222
+)
+
+
+@app.listen("store.listen")
+async def listen(message):
+    try:
+        print(message.subject, message.data)
+    except Exception as ex:
+        app.logger.exception(message.subject)
+
+
+@app.listen("store.request")
+async def response(message):
+    try:
+        print(message.subject, message.data)
+        return {"data": "request"}
+    except Exception as ex:
+        app.logger.exception(message.subject)
+
+
+emulator = EmulatorClient('resources/events.writer.2021-03-16-13:06.jsonl', emulate_timeout=False)
+app.add_middleware(ReaderEmulatorMiddleware)
+
+if __name__ == "__main__":
+
+    print("app")
+    start_process(app.start)
+    time.sleep(4)
+    print("emulator")
+    asyncio.new_event_loop().run_until_complete(emulator.run())
+
+    print("wait")
+    asyncio.new_event_loop().run_until_complete(emulator.wait())
+    print("done")
diff --git a/examples/emulator/resources/events.listener.2021-03-19-16:21:13.jsonl b/examples/emulator/resources/events.listener.2021-03-19-16:21:13.jsonl
new file mode 100644
index 0000000..2fe7d81
--- /dev/null
+++ b/examples/emulator/resources/events.listener.2021-03-19-16:21:13.jsonl
@@ -0,0 +1,20 @@
+{"event_type": "listen_request", "subject": "listener.store.request", "message": {"data": "request.data.0"}, "context": {}, "response": {"data": "request"}, "timestamp": 1616163690.085264}
+{"event_type": "listen_publish", "subject": "listener.store.listen", "message": {"data": "publish.data.0"}, "context": {}, "timestamp": 1616163690.085299}
+{"event_type": "listen_publish", "subject": "listener.store.listen", "message": {"data": "publish.data.1"}, "context": {}, "timestamp": 1616163691.0911539}
+{"event_type": "listen_request", "subject": "listener.store.request", "message": {"data": "request.data.1"}, "context": {}, "response": {"data": "request"}, "timestamp": 1616163691.594288}
+{"event_type": "listen_publish", "subject": "listener.store.listen", "message": {"data": "publish.data.2"}, "context": {}, "timestamp": 1616163692.0904489}
+{"event_type": "listen_publish", "subject": "listener.store.listen", "message": {"data": "publish.data.3"}, "context": {}, "timestamp": 1616163693.095857}
+{"event_type": "listen_request", "subject": "listener.store.request", "message": {"data": "request.data.2"}, "context": {}, "response": {"data": "request"}, "timestamp": 1616163693.101412}
+{"event_type": "listen_publish", "subject": "listener.store.listen", "message": {"data": "publish.data.4"}, "context": {}, "timestamp": 1616163694.0976908}
+{"event_type": "listen_request", "subject": "listener.store.request", "message": {"data": "request.data.3"}, "context": {}, "response": {"data": "request"}, "timestamp": 1616163694.611608}
+{"event_type": "listen_publish", "subject": "listener.store.listen", "message": {"data": "publish.data.5"}, "context": {}, "timestamp": 1616163695.100329}
+{"event_type": "listen_publish", "subject": "listener.store.listen", "message": {"data": "publish.data.6"}, "context": {}, "timestamp": 1616163696.104001}
+{"event_type": "listen_request", "subject": "listener.store.request", "message": {"data": "request.data.4"}, "context": {}, "response": {"data": "request"}, "timestamp": 1616163696.121866}
+{"event_type": "listen_publish", "subject": "listener.store.listen", "message": {"data": "publish.data.7"}, "context": {}, "timestamp": 1616163697.106553}
+{"event_type": "listen_request", "subject": "listener.store.request", "message": {"data": "request.data.5"}, "context": {}, "response": {"data": "request"}, "timestamp": 1616163697.6313019}
+{"event_type": "listen_publish", "subject": "listener.store.listen", "message": {"data": "publish.data.8"}, "context": {}, "timestamp": 1616163698.1104178}
+{"event_type": "listen_publish", "subject": "listener.store.listen", "message": {"data": "publish.data.9"}, "context": {}, "timestamp": 1616163699.1148028}
+{"event_type": "listen_request", "subject": "listener.store.request", "message": {"data": "request.data.6"}, "context": {}, "response": {"data": "request"}, "timestamp": 1616163699.143626}
+{"event_type": "listen_request", "subject": "listener.store.request", "message": {"data": "request.data.7"}, "context": {}, "response": {"data": "request"}, "timestamp": 1616163700.654367}
+{"event_type": "listen_request", "subject": "listener.store.request", "message": {"data": "request.data.8"}, "context": {}, "response": {"data": "request"}, "timestamp": 1616163702.167836}
+{"event_type": "listen_request", "subject": "listener.store.request", "message": {"data": "request.data.9"}, "context": {}, "response": {"data": "request"}, "timestamp": 1616163703.6755}
diff --git a/examples/emulator/resources/events.listener_publisher.2021-03-19-16:22:26.jsonl b/examples/emulator/resources/events.listener_publisher.2021-03-19-16:22:26.jsonl
new file mode 100644
index 0000000..5c42d7b
--- /dev/null
+++ b/examples/emulator/resources/events.listener_publisher.2021-03-19-16:22:26.jsonl
@@ -0,0 +1,40 @@
+{"event_type": "send_publish", "subject": "store.listen", "message": {"data": "publish.data.0"}, "timestamp": 1616163746.40906}
+{"event_type": "listen_request", "subject": "store.request", "message": {"data": "request.data.0"}, "context": {}, "response": {"data": "request"}, "timestamp": 1616163746.411051}
+{"event_type": "listen_publish", "subject": "store.listen", "message": {"data": "publish.data.0"}, "context": {}, "timestamp": 1616163746.411084}
+{"event_type": "send_request", "subject": "store.request", "message": {"data": "request.data.0"}, "response": {"data": "request"}, "timestamp": 1616163746.4127538}
+{"event_type": "send_publish", "subject": "store.listen", "message": {"data": "publish.data.1"}, "timestamp": 1616163747.413412}
+{"event_type": "listen_publish", "subject": "store.listen", "message": {"data": "publish.data.1"}, "context": {}, "timestamp": 1616163747.415643}
+{"event_type": "listen_request", "subject": "store.request", "message": {"data": "request.data.1"}, "context": {}, "response": {"data": "request"}, "timestamp": 1616163747.918144}
+{"event_type": "send_request", "subject": "store.request", "message": {"data": "request.data.1"}, "response": {"data": "request"}, "timestamp": 1616163747.920809}
+{"event_type": "send_publish", "subject": "store.listen", "message": {"data": "publish.data.2"}, "timestamp": 1616163748.4150531}
+{"event_type": "listen_publish", "subject": "store.listen", "message": {"data": "publish.data.2"}, "context": {}, "timestamp": 1616163748.417831}
+{"event_type": "send_publish", "subject": "store.listen", "message": {"data": "publish.data.3"}, "timestamp": 1616163749.418966}
+{"event_type": "listen_publish", "subject": "store.listen", "message": {"data": "publish.data.3"}, "context": {}, "timestamp": 1616163749.42243}
+{"event_type": "listen_request", "subject": "store.request", "message": {"data": "request.data.2"}, "context": {}, "response": {"data": "request"}, "timestamp": 1616163749.424376}
+{"event_type": "send_request", "subject": "store.request", "message": {"data": "request.data.2"}, "response": {"data": "request"}, "timestamp": 1616163749.4262688}
+{"event_type": "send_publish", "subject": "store.listen", "message": {"data": "publish.data.4"}, "timestamp": 1616163750.423672}
+{"event_type": "listen_publish", "subject": "store.listen", "message": {"data": "publish.data.4"}, "context": {}, "timestamp": 1616163750.428416}
+{"event_type": "listen_request", "subject": "store.request", "message": {"data": "request.data.3"}, "context": {}, "response": {"data": "request"}, "timestamp": 1616163750.933328}
+{"event_type": "send_request", "subject": "store.request", "message": {"data": "request.data.3"}, "response": {"data": "request"}, "timestamp": 1616163750.935594}
+{"event_type": "send_publish", "subject": "store.listen", "message": {"data": "publish.data.5"}, "timestamp": 1616163751.426971}
+{"event_type": "listen_publish", "subject": "store.listen", "message": {"data": "publish.data.5"}, "context": {}, "timestamp": 1616163751.4310882}
+{"event_type": "send_publish", "subject": "store.listen", "message": {"data": "publish.data.6"}, "timestamp": 1616163752.4306839}
+{"event_type": "listen_publish", "subject": "store.listen", "message": {"data": "publish.data.6"}, "context": {}, "timestamp": 1616163752.433504}
+{"event_type": "listen_request", "subject": "store.request", "message": {"data": "request.data.4"}, "context": {}, "response": {"data": "request"}, "timestamp": 1616163752.4387121}
+{"event_type": "send_request", "subject": "store.request", "message": {"data": "request.data.4"}, "response": {"data": "request"}, "timestamp": 1616163752.440761}
+{"event_type": "send_publish", "subject": "store.listen", "message": {"data": "publish.data.7"}, "timestamp": 1616163753.4312801}
+{"event_type": "listen_publish", "subject": "store.listen", "message": {"data": "publish.data.7"}, "context": {}, "timestamp": 1616163753.434462}
+{"event_type": "listen_request", "subject": "store.request", "message": {"data": "request.data.5"}, "context": {}, "response": {"data": "request"}, "timestamp": 1616163753.948512}
+{"event_type": "send_request", "subject": "store.request", "message": {"data": "request.data.5"}, "response": {"data": "request"}, "timestamp": 1616163753.951016}
+{"event_type": "send_publish", "subject": "store.listen", "message": {"data": "publish.data.8"}, "timestamp": 1616163754.4349}
+{"event_type": "listen_publish", "subject": "store.listen", "message": {"data": "publish.data.8"}, "context": {}, "timestamp": 1616163754.4377851}
+{"event_type": "send_publish", "subject": "store.listen", "message": {"data": "publish.data.9"}, "timestamp": 1616163755.439557}
+{"event_type": "listen_publish", "subject": "store.listen", "message": {"data": "publish.data.9"}, "context": {}, "timestamp": 1616163755.4424572}
+{"event_type": "listen_request", "subject": "store.request", "message": {"data": "request.data.6"}, "context": {}, "response": {"data": "request"}, "timestamp": 1616163755.455765}
+{"event_type": "send_request", "subject": "store.request", "message": {"data": "request.data.6"}, "response": {"data": "request"}, "timestamp": 1616163755.458336}
+{"event_type": "listen_request", "subject": "store.request", "message": {"data": "request.data.7"}, "context": {}, "response": {"data": "request"}, "timestamp": 1616163756.966461}
+{"event_type": "send_request", "subject": "store.request", "message": {"data": "request.data.7"}, "response": {"data": "request"}, "timestamp": 1616163756.968477}
+{"event_type": "listen_request", "subject": "store.request", "message": {"data": "request.data.8"}, "context": {}, "response": {"data": "request"}, "timestamp": 1616163758.4733608}
+{"event_type": "send_request", "subject": "store.request", "message": {"data": "request.data.8"}, "response": {"data": "request"}, "timestamp": 1616163758.475225}
+{"event_type": "listen_request", "subject": "store.request", "message": {"data": "request.data.9"}, "context": {}, "response": {"data": "request"}, "timestamp": 1616163759.9795651}
+{"event_type": "send_request", "subject": "store.request", "message": {"data": "request.data.9"}, "response": {"data": "request"}, "timestamp": 1616163759.981707}
diff --git a/examples/emulator/resources/events.listener_publisher.2021-03-19-19:58:06.jsonl b/examples/emulator/resources/events.listener_publisher.2021-03-19-19:58:06.jsonl
new file mode 100644
index 0000000..300a04e
--- /dev/null
+++ b/examples/emulator/resources/events.listener_publisher.2021-03-19-19:58:06.jsonl
@@ -0,0 +1,31 @@
+{"event_type": "send_publish", "subject": "panini_events.listener_publisher.listener_publisher__non_docker_env_649030__659046.started", "message": {}, "timestamp": 1616176686.77689}
+{"event_type": "send_publish", "subject": "store.listen", "message": {"data": "publish.data.0"}, "timestamp": 1616176686.7773812}
+{"event_type": "listen_request", "subject": "store.request", "message": {"data": "request.data.0"}, "context": {}, "response": {"data": "request"}, "timestamp": 1616176686.778959}
+{"event_type": "listen_publish", "subject": "store.listen", "message": {"data": "publish.data.0"}, "context": {}, "timestamp": 1616176686.7790248}
+{"event_type": "send_request", "subject": "store.request", "message": {"data": "request.data.0"}, "response": {"data": "request"}, "timestamp": 1616176686.7810292}
+{"event_type": "send_publish", "subject": "store.listen", "message": {"data": "publish.data.1"}, "timestamp": 1616176687.781695}
+{"event_type": "listen_publish", "subject": "store.listen", "message": {"data": "publish.data.1"}, "context": {}, "timestamp": 1616176687.784008}
+{"event_type": "listen_request", "subject": "store.request", "message": {"data": "request.data.1"}, "context": {}, "response": {"data": "request"}, "timestamp": 1616176688.288361}
+{"event_type": "send_request", "subject": "store.request", "message": {"data": "request.data.1"}, "response": {"data": "request"}, "timestamp": 1616176688.290153}
+{"event_type": "send_publish", "subject": "store.listen", "message": {"data": "publish.data.2"}, "timestamp": 1616176688.784297}
+{"event_type": "listen_publish", "subject": "store.listen", "message": {"data": "publish.data.2"}, "context": {}, "timestamp": 1616176688.787374}
+{"event_type": "send_publish", "subject": "store.listen", "message": {"data": "publish.data.3"}, "timestamp": 1616176689.7869911}
+{"event_type": "listen_publish", "subject": "store.listen", "message": {"data": "publish.data.3"}, "context": {}, "timestamp": 1616176689.7897978}
+{"event_type": "listen_request", "subject": "store.request", "message": {"data": "request.data.2"}, "context": {}, "response": {"data": "request"}, "timestamp": 1616176689.792532}
+{"event_type": "send_request", "subject": "store.request", "message": {"data": "request.data.2"}, "response": {"data": "request"}, "timestamp": 1616176689.794294}
+{"event_type": "send_publish", "subject": "store.listen", "message": {"data": "publish.data.4"}, "timestamp": 1616176690.78984}
+{"event_type": "listen_publish", "subject": "store.listen", "message": {"data": "publish.data.4"}, "context": {}, "timestamp": 1616176690.792812}
+{"event_type": "listen_request", "subject": "store.request", "message": {"data": "request.data.3"}, "context": {}, "response": {"data": "request"}, "timestamp": 1616176691.302924}
+{"event_type": "send_request", "subject": "store.request", "message": {"data": "request.data.3"}, "response": {"data": "request"}, "timestamp": 1616176691.305665}
+{"event_type": "send_publish", "subject": "store.listen", "message": {"data": "publish.data.5"}, "timestamp": 1616176691.794452}
+{"event_type": "listen_publish", "subject": "store.listen", "message": {"data": "publish.data.5"}, "context": {}, "timestamp": 1616176691.796705}
+{"event_type": "send_publish", "subject": "store.listen", "message": {"data": "publish.data.6"}, "timestamp": 1616176692.794969}
+{"event_type": "listen_publish", "subject": "store.listen", "message": {"data": "publish.data.6"}, "context": {}, "timestamp": 1616176692.7976089}
+{"event_type": "listen_request", "subject": "store.request", "message": {"data": "request.data.4"}, "context": {}, "response": {"data": "request"}, "timestamp": 1616176692.81003}
+{"event_type": "send_request", "subject": "store.request", "message": {"data": "request.data.4"}, "response": {"data": "request"}, "timestamp": 1616176692.8121521}
+{"event_type": "send_publish", "subject": "store.listen", "message": {"data": "publish.data.7"}, "timestamp": 1616176693.797626}
+{"event_type": "listen_publish", "subject": "store.listen", "message": {"data": "publish.data.7"}, "context": {}, "timestamp": 1616176693.8000078}
+{"event_type": "listen_request", "subject": "store.request", "message": {"data": "request.data.5"}, "context": {}, "response": {"data": "request"}, "timestamp": 1616176694.317716}
+{"event_type": "send_request", "subject": "store.request", "message": {"data": "request.data.5"}, "response": {"data": "request"}, "timestamp": 1616176694.320859}
+{"event_type": "send_publish", "subject": "store.listen", "message": {"data": "publish.data.8"}, "timestamp": 1616176694.800283}
+{"event_type": "listen_publish", "subject": "store.listen", "message": {"data": "publish.data.8"}, "context": {}, "timestamp": 1616176694.802691}
diff --git a/examples/emulator/resources/events.publisher.2021-03-19-16:21:30.jsonl b/examples/emulator/resources/events.publisher.2021-03-19-16:21:30.jsonl
new file mode 100644
index 0000000..e631e42
--- /dev/null
+++ b/examples/emulator/resources/events.publisher.2021-03-19-16:21:30.jsonl
@@ -0,0 +1,20 @@
+{"event_type": "send_publish", "subject": "listener.store.listen", "message": {"data": "publish.data.0"}, "timestamp": 1616163690.083235}
+{"event_type": "send_request", "subject": "listener.store.request", "message": {"data": "request.data.0"}, "response": {"data": "request"}, "timestamp": 1616163690.087649}
+{"event_type": "send_publish", "subject": "listener.store.listen", "message": {"data": "publish.data.1"}, "timestamp": 1616163691.0876448}
+{"event_type": "send_request", "subject": "listener.store.request", "message": {"data": "request.data.1"}, "response": {"data": "request"}, "timestamp": 1616163691.597118}
+{"event_type": "send_publish", "subject": "listener.store.listen", "message": {"data": "publish.data.2"}, "timestamp": 1616163692.088084}
+{"event_type": "send_publish", "subject": "listener.store.listen", "message": {"data": "publish.data.3"}, "timestamp": 1616163693.092993}
+{"event_type": "send_request", "subject": "listener.store.request", "message": {"data": "request.data.2"}, "response": {"data": "request"}, "timestamp": 1616163693.103914}
+{"event_type": "send_publish", "subject": "listener.store.listen", "message": {"data": "publish.data.4"}, "timestamp": 1616163694.094548}
+{"event_type": "send_request", "subject": "listener.store.request", "message": {"data": "request.data.3"}, "response": {"data": "request"}, "timestamp": 1616163694.616678}
+{"event_type": "send_publish", "subject": "listener.store.listen", "message": {"data": "publish.data.5"}, "timestamp": 1616163695.0966752}
+{"event_type": "send_publish", "subject": "listener.store.listen", "message": {"data": "publish.data.6"}, "timestamp": 1616163696.100922}
+{"event_type": "send_request", "subject": "listener.store.request", "message": {"data": "request.data.4"}, "response": {"data": "request"}, "timestamp": 1616163696.124633}
+{"event_type": "send_publish", "subject": "listener.store.listen", "message": {"data": "publish.data.7"}, "timestamp": 1616163697.1034791}
+{"event_type": "send_request", "subject": "listener.store.request", "message": {"data": "request.data.5"}, "response": {"data": "request"}, "timestamp": 1616163697.634209}
+{"event_type": "send_publish", "subject": "listener.store.listen", "message": {"data": "publish.data.8"}, "timestamp": 1616163698.1078699}
+{"event_type": "send_publish", "subject": "listener.store.listen", "message": {"data": "publish.data.9"}, "timestamp": 1616163699.111933}
+{"event_type": "send_request", "subject": "listener.store.request", "message": {"data": "request.data.6"}, "response": {"data": "request"}, "timestamp": 1616163699.146724}
+{"event_type": "send_request", "subject": "listener.store.request", "message": {"data": "request.data.7"}, "response": {"data": "request"}, "timestamp": 1616163700.6579158}
+{"event_type": "send_request", "subject": "listener.store.request", "message": {"data": "request.data.8"}, "response": {"data": "request"}, "timestamp": 1616163702.170418}
+{"event_type": "send_request", "subject": "listener.store.request", "message": {"data": "request.data.9"}, "response": {"data": "request"}, "timestamp": 1616163703.677749}
diff --git a/examples/emulator/writer.py b/examples/emulator/writer.py
new file mode 100644
index 0000000..38b0a59
--- /dev/null
+++ b/examples/emulator/writer.py
@@ -0,0 +1,55 @@
+import asyncio
+from datetime import datetime
+
+from panini.emulator.writer_emulator_middleware import WriterEmulatorMiddleware
+from panini.app import App
+
+app = App(
+    service_name="writer",
+    host="127.0.0.1",
+    port=4222
+)
+
+
+@app.listen("store.listen")
+async def listen(message):
+    try:
+        pass
+    except Exception as ex:
+        app.logger.exception(message.subject)
+
+
+@app.listen("store.request")
+async def response(message):
+    try:
+        return {"data": "request"}
+    except Exception as ex:
+        app.logger.exception(message.subject)
+
+
+@app.task()
+async def request_task():
+    for i in range(10):
+        subject = "store.request"
+        try:
+            await app.request(subject, {"data": f"request.data.{i}"})
+            await asyncio.sleep(3)
+        except Exception as ex:
+            app.logger.exception(subject)
+    print("finish request")
+
+@app.task()
+async def publish_task():
+    for i in range(10):
+        subject = "store.listen"
+        try:
+            await app.publish(subject, {"data": f"publish.data.{i}"})
+            await asyncio.sleep(2)
+        except Exception as ex:
+            app.logger.exception(subject)
+    print("finish publish")
+
+if __name__ == "__main__":
+    events_filename = f"resources/events.{app.service_name}.{datetime.now().strftime('%Y-%m-%d-%H:%M')}.jsonl"
+    app.add_middleware(WriterEmulatorMiddleware, filename=events_filename)
+    app.start()
diff --git a/examples/simple_examples/async_middleware.py b/examples/simple_examples/async_middleware.py
index 2e18d9e..a527f99 100644
--- a/examples/simple_examples/async_middleware.py
+++ b/examples/simple_examples/async_middleware.py
@@ -47,9 +47,9 @@ async def receive_messages(msg):
 
 
 class MyMiddleware(Middleware):
-    async def send_publish(self, subject, message, publish_func, **kwargs):
+    async def send_publish(self, subject, message, publish_func, *args, **kwargs):
         print("do something before publish")
-        await publish_func(subject, message, **kwargs)
+        await publish_func(subject, message, *args, **kwargs)
         print("do something after publish")
 
     async def listen_publish(self, msg, cb):
@@ -57,9 +57,9 @@ async def listen_publish(self, msg, cb):
         await cb(msg)
         print("do something after listen")
 
-    async def send_request(self, subject, message, request_func, **kwargs):
+    async def send_request(self, subject, message, request_func, *args, **kwargs):
         print("do something before send request")
-        result = await request_func(subject, message, **kwargs)
+        result = await request_func(subject, message, *args, **kwargs)
         print("do something after send request")
         return result
 
diff --git a/grafana_dashboard/panini_dashboard.json b/grafana_dashboard/panini_dashboard.json
new file mode 100644
index 0000000..7e12a33
--- /dev/null
+++ b/grafana_dashboard/panini_dashboard.json
@@ -0,0 +1,691 @@
+{
+  "annotations": {
+    "list": [
+      {
+        "builtIn": 1,
+        "datasource": "-- Grafana --",
+        "enable": true,
+        "hide": true,
+        "iconColor": "rgba(0, 211, 255, 1)",
+        "name": "Annotations & Alerts",
+        "type": "dashboard"
+      }
+    ]
+  },
+  "editable": true,
+  "gnetId": null,
+  "graphTooltip": 0,
+  "iteration": 1616590977569,
+  "links": [],
+  "panels": [
+    {
+      "collapsed": false,
+      "datasource": null,
+      "gridPos": {
+        "h": 1,
+        "w": 24,
+        "x": 0,
+        "y": 0
+      },
+      "id": 14,
+      "panels": [],
+      "title": "Global",
+      "type": "row"
+    },
+    {
+      "datasource": "Prometheus",
+      "fieldConfig": {
+        "defaults": {
+          "custom": {},
+          "mappings": [],
+          "thresholds": {
+            "mode": "absolute",
+            "steps": [
+              {
+                "color": "green",
+                "value": null
+              },
+              {
+                "color": "red",
+                "value": 80
+              }
+            ]
+          }
+        },
+        "overrides": []
+      },
+      "gridPos": {
+        "h": 2,
+        "w": 12,
+        "x": 0,
+        "y": 1
+      },
+      "id": 10,
+      "options": {
+        "colorMode": "value",
+        "graphMode": "area",
+        "justifyMode": "auto",
+        "orientation": "auto",
+        "reduceOptions": {
+          "calcs": [
+            "mean"
+          ],
+          "fields": "",
+          "values": false
+        },
+        "textMode": "auto"
+      },
+      "pluginVersion": "7.3.1",
+      "targets": [
+        {
+          "expr": "sum(irate(panini_listen_total{app_name=~\"^$app_name$\"}[1m]))",
+          "interval": "",
+          "legendFormat": "",
+          "refId": "A"
+        }
+      ],
+      "timeFrom": null,
+      "timeShift": null,
+      "title": "Total requests",
+      "type": "stat"
+    },
+    {
+      "datasource": "Prometheus",
+      "fieldConfig": {
+        "defaults": {
+          "custom": {},
+          "mappings": [],
+          "thresholds": {
+            "mode": "absolute",
+            "steps": [
+              {
+                "color": "green",
+                "value": null
+              },
+              {
+                "color": "red",
+                "value": 80
+              }
+            ]
+          }
+        },
+        "overrides": []
+      },
+      "gridPos": {
+        "h": 2,
+        "w": 6,
+        "x": 12,
+        "y": 1
+      },
+      "id": 11,
+      "options": {
+        "colorMode": "value",
+        "graphMode": "area",
+        "justifyMode": "auto",
+        "orientation": "auto",
+        "reduceOptions": {
+          "calcs": [
+            "mean"
+          ],
+          "fields": "",
+          "values": false
+        },
+        "textMode": "auto"
+      },
+      "pluginVersion": "7.3.1",
+      "targets": [
+        {
+          "expr": "sum(irate(panini_listen_total{status=\"success\", app_name=~\"^$app_name$\"}[1m]))",
+          "interval": "",
+          "legendFormat": "",
+          "refId": "A"
+        }
+      ],
+      "timeFrom": null,
+      "timeShift": null,
+      "title": "Successful requests",
+      "type": "stat"
+    },
+    {
+      "datasource": "Prometheus",
+      "fieldConfig": {
+        "defaults": {
+          "custom": {},
+          "mappings": [],
+          "thresholds": {
+            "mode": "absolute",
+            "steps": [
+              {
+                "color": "green",
+                "value": null
+              },
+              {
+                "color": "red",
+                "value": 80
+              }
+            ]
+          }
+        },
+        "overrides": []
+      },
+      "gridPos": {
+        "h": 2,
+        "w": 6,
+        "x": 18,
+        "y": 1
+      },
+      "id": 12,
+      "options": {
+        "colorMode": "value",
+        "graphMode": "area",
+        "justifyMode": "auto",
+        "orientation": "auto",
+        "reduceOptions": {
+          "calcs": [
+            "mean"
+          ],
+          "fields": "",
+          "values": false
+        },
+        "textMode": "auto"
+      },
+      "pluginVersion": "7.3.1",
+      "targets": [
+        {
+          "expr": "sum(irate(panini_listen_total{status=\"failed\", app_name=~\"^$app_name$\"}[1m]))",
+          "interval": "",
+          "legendFormat": "",
+          "refId": "A"
+        }
+      ],
+      "timeFrom": null,
+      "timeShift": null,
+      "title": "Failed requests",
+      "type": "stat"
+    },
+    {
+      "collapsed": false,
+      "datasource": null,
+      "gridPos": {
+        "h": 1,
+        "w": 24,
+        "x": 0,
+        "y": 3
+      },
+      "id": 8,
+      "panels": [],
+      "title": "Received messages",
+      "type": "row"
+    },
+    {
+      "aliasColors": {},
+      "bars": false,
+      "dashLength": 10,
+      "dashes": false,
+      "datasource": "Prometheus",
+      "fieldConfig": {
+        "defaults": {
+          "custom": {}
+        },
+        "overrides": []
+      },
+      "fill": 1,
+      "fillGradient": 0,
+      "gridPos": {
+        "h": 7,
+        "w": 12,
+        "x": 0,
+        "y": 4
+      },
+      "hiddenSeries": false,
+      "id": 6,
+      "legend": {
+        "avg": false,
+        "current": false,
+        "max": false,
+        "min": false,
+        "show": true,
+        "total": false,
+        "values": false
+      },
+      "lines": true,
+      "linewidth": 1,
+      "nullPointMode": "null",
+      "options": {
+        "alertThreshold": true
+      },
+      "percentage": false,
+      "pluginVersion": "7.3.1",
+      "pointradius": 2,
+      "points": false,
+      "renderer": "flot",
+      "seriesOverrides": [],
+      "spaceLength": 10,
+      "stack": false,
+      "steppedLine": false,
+      "targets": [
+        {
+          "expr": "sum(rate(panini_listen_total{app_name=~\"^$app_name$\"}[15m]))by (subject)",
+          "interval": "",
+          "legendFormat": "{{subject}}",
+          "refId": "A"
+        }
+      ],
+      "thresholds": [],
+      "timeFrom": null,
+      "timeRegions": [],
+      "timeShift": null,
+      "title": "Msg rate by subject",
+      "tooltip": {
+        "shared": true,
+        "sort": 0,
+        "value_type": "individual"
+      },
+      "type": "graph",
+      "xaxis": {
+        "buckets": null,
+        "mode": "time",
+        "name": null,
+        "show": true,
+        "values": []
+      },
+      "yaxes": [
+        {
+          "format": "short",
+          "label": null,
+          "logBase": 1,
+          "max": null,
+          "min": null,
+          "show": true
+        },
+        {
+          "format": "short",
+          "label": null,
+          "logBase": 1,
+          "max": null,
+          "min": null,
+          "show": true
+        }
+      ],
+      "yaxis": {
+        "align": false,
+        "alignLevel": null
+      }
+    },
+    {
+      "aliasColors": {},
+      "bars": false,
+      "dashLength": 10,
+      "dashes": false,
+      "datasource": "Prometheus",
+      "fieldConfig": {
+        "defaults": {
+          "custom": {}
+        },
+        "overrides": []
+      },
+      "fill": 1,
+      "fillGradient": 0,
+      "gridPos": {
+        "h": 7,
+        "w": 12,
+        "x": 12,
+        "y": 4
+      },
+      "hiddenSeries": false,
+      "id": 2,
+      "legend": {
+        "avg": false,
+        "current": false,
+        "max": false,
+        "min": false,
+        "show": true,
+        "total": false,
+        "values": false
+      },
+      "lines": true,
+      "linewidth": 1,
+      "nullPointMode": "null",
+      "options": {
+        "alertThreshold": true
+      },
+      "percentage": false,
+      "pluginVersion": "7.3.1",
+      "pointradius": 2,
+      "points": false,
+      "renderer": "flot",
+      "seriesOverrides": [],
+      "spaceLength": 10,
+      "stack": false,
+      "steppedLine": false,
+      "targets": [
+        {
+          "expr": "sum(rate(panini_listen_total{app_name=~\"^$app_name$\"}[15m])) by (status, app_name)",
+          "interval": "",
+          "legendFormat": "{{status}}",
+          "refId": "A"
+        }
+      ],
+      "thresholds": [],
+      "timeFrom": null,
+      "timeRegions": [],
+      "timeShift": null,
+      "title": "Msg rate by status",
+      "tooltip": {
+        "shared": true,
+        "sort": 0,
+        "value_type": "individual"
+      },
+      "type": "graph",
+      "xaxis": {
+        "buckets": null,
+        "mode": "time",
+        "name": null,
+        "show": true,
+        "values": []
+      },
+      "yaxes": [
+        {
+          "format": "short",
+          "label": null,
+          "logBase": 1,
+          "max": null,
+          "min": null,
+          "show": true
+        },
+        {
+          "format": "short",
+          "label": null,
+          "logBase": 1,
+          "max": null,
+          "min": null,
+          "show": true
+        }
+      ],
+      "yaxis": {
+        "align": false,
+        "alignLevel": null
+      }
+    },
+    {
+      "aliasColors": {},
+      "bars": false,
+      "dashLength": 10,
+      "dashes": false,
+      "datasource": "Prometheus",
+      "fieldConfig": {
+        "defaults": {
+          "custom": {}
+        },
+        "overrides": []
+      },
+      "fill": 1,
+      "fillGradient": 0,
+      "gridPos": {
+        "h": 6,
+        "w": 12,
+        "x": 0,
+        "y": 11
+      },
+      "hiddenSeries": false,
+      "id": 3,
+      "legend": {
+        "avg": false,
+        "current": false,
+        "max": false,
+        "min": false,
+        "show": true,
+        "total": false,
+        "values": false
+      },
+      "lines": true,
+      "linewidth": 1,
+      "nullPointMode": "null",
+      "options": {
+        "alertThreshold": true
+      },
+      "percentage": false,
+      "pluginVersion": "7.3.1",
+      "pointradius": 2,
+      "points": false,
+      "renderer": "flot",
+      "seriesOverrides": [],
+      "spaceLength": 10,
+      "stack": false,
+      "steppedLine": false,
+      "targets": [
+        {
+          "expr": "histogram_quantile(0.99, sum(rate(panini_listen_latency_seconds_bucket{app_name=~\"^$app_name$\"}[15m])) by (le, app_name, subject))",
+          "interval": "",
+          "legendFormat": "{{subject}}",
+          "refId": "A"
+        }
+      ],
+      "thresholds": [],
+      "timeFrom": null,
+      "timeRegions": [],
+      "timeShift": null,
+      "title": "Msg latency by subject",
+      "tooltip": {
+        "shared": true,
+        "sort": 0,
+        "value_type": "individual"
+      },
+      "type": "graph",
+      "xaxis": {
+        "buckets": null,
+        "mode": "time",
+        "name": null,
+        "show": true,
+        "values": []
+      },
+      "yaxes": [
+        {
+          "format": "short",
+          "label": null,
+          "logBase": 1,
+          "max": null,
+          "min": null,
+          "show": true
+        },
+        {
+          "format": "short",
+          "label": null,
+          "logBase": 1,
+          "max": null,
+          "min": null,
+          "show": true
+        }
+      ],
+      "yaxis": {
+        "align": false,
+        "alignLevel": null
+      }
+    },
+    {
+      "aliasColors": {},
+      "bars": false,
+      "dashLength": 10,
+      "dashes": false,
+      "datasource": "Prometheus",
+      "fieldConfig": {
+        "defaults": {
+          "custom": {}
+        },
+        "overrides": []
+      },
+      "fill": 1,
+      "fillGradient": 0,
+      "gridPos": {
+        "h": 6,
+        "w": 12,
+        "x": 12,
+        "y": 11
+      },
+      "hiddenSeries": false,
+      "id": 5,
+      "legend": {
+        "avg": false,
+        "current": false,
+        "max": false,
+        "min": false,
+        "show": true,
+        "total": false,
+        "values": false
+      },
+      "lines": true,
+      "linewidth": 1,
+      "nullPointMode": "null",
+      "options": {
+        "alertThreshold": true
+      },
+      "percentage": false,
+      "pluginVersion": "7.3.1",
+      "pointradius": 2,
+      "points": false,
+      "renderer": "flot",
+      "seriesOverrides": [],
+      "spaceLength": 10,
+      "stack": false,
+      "steppedLine": false,
+      "targets": [
+        {
+          "expr": "histogram_quantile(0.99, sum(rate(panini_listen_latency_seconds_bucket{app_name=~\"^$app_name$\"}[5m])) by (le, app_name))",
+          "interval": "",
+          "legendFormat": "p=0.99",
+          "refId": "A"
+        },
+        {
+          "expr": "histogram_quantile(0.95, sum(rate(panini_listen_latency_seconds_bucket{app_name=~\"^$app_name$\"}[5m])) by (le, app_name))",
+          "interval": "",
+          "legendFormat": "p=95",
+          "refId": "B"
+        },
+        {
+          "expr": "histogram_quantile(0.90, sum(rate(panini_listen_latency_seconds_bucket{app_name=~\"^$app_name$\"}[5m])) by (le, app_name))",
+          "interval": "",
+          "legendFormat": "p=0.90",
+          "refId": "C"
+        },
+        {
+          "expr": "histogram_quantile(0.50, sum(rate(panini_listen_latency_seconds_bucket{app_name=~\"^$app_name$\"}[5m])) by (le, app_name))",
+          "interval": "",
+          "legendFormat": "p=0.50",
+          "refId": "D"
+        }
+      ],
+      "thresholds": [],
+      "timeFrom": null,
+      "timeRegions": [],
+      "timeShift": null,
+      "title": "Msg latency quantile",
+      "tooltip": {
+        "shared": true,
+        "sort": 0,
+        "value_type": "individual"
+      },
+      "type": "graph",
+      "xaxis": {
+        "buckets": null,
+        "mode": "time",
+        "name": null,
+        "show": true,
+        "values": []
+      },
+      "yaxes": [
+        {
+          "format": "short",
+          "label": null,
+          "logBase": 1,
+          "max": null,
+          "min": null,
+          "show": true
+        },
+        {
+          "format": "short",
+          "label": null,
+          "logBase": 1,
+          "max": null,
+          "min": null,
+          "show": true
+        }
+      ],
+      "yaxis": {
+        "align": false,
+        "alignLevel": null
+      }
+    }
+  ],
+  "refresh": "5s",
+  "schemaVersion": 26,
+  "style": "dark",
+  "tags": [],
+  "templating": {
+    "list": [
+      {
+        "allValue": null,
+        "current": {
+          "selected": false,
+          "text": "wss_NATS_listener",
+          "value": "wss_NATS_listener"
+        },
+        "datasource": "Prometheus",
+        "definition": "label_values(panini_listen_latency_seconds_bucket, app_name)",
+        "error": null,
+        "hide": 0,
+        "includeAll": false,
+        "label": "app",
+        "multi": false,
+        "name": "app_name",
+        "options": [],
+        "query": "label_values(panini_listen_latency_seconds_bucket, app_name)",
+        "refresh": 2,
+        "regex": "",
+        "skipUrlSync": false,
+        "sort": 0,
+        "tagValuesQuery": "",
+        "tags": [],
+        "tagsQuery": "",
+        "type": "query",
+        "useTags": false
+      },
+      {
+        "allValue": null,
+        "current": {
+          "isNone": true,
+          "selected": false,
+          "text": "None",
+          "value": ""
+        },
+        "datasource": "Prometheus",
+        "definition": "label_values(panini_listen_latency_seconds_bucket, server)",
+        "error": null,
+        "hide": 0,
+        "includeAll": false,
+        "label": "server",
+        "multi": false,
+        "name": "server",
+        "options": [],
+        "query": "label_values(panini_listen_latency_seconds_bucket, server)",
+        "refresh": 1,
+        "regex": "",
+        "skipUrlSync": false,
+        "sort": 0,
+        "tagValuesQuery": "",
+        "tags": [],
+        "tagsQuery": "",
+        "type": "query",
+        "useTags": false
+      }
+    ]
+  },
+  "time": {
+    "from": "now-6h",
+    "to": "now"
+  },
+  "timepicker": {},
+  "timezone": "",
+  "title": "Panini microservice",
+  "uid": "U-8PwmQMz",
+  "version": 4
+}
\ No newline at end of file
diff --git a/panini/app.py b/panini/app.py
index 3b3f92c..6ff3ce3 100644
--- a/panini/app.py
+++ b/panini/app.py
@@ -95,7 +95,7 @@ def __init__(
             else:
                 client_id = client_id
             os.environ["CLIENT_ID"] = client_id
-            self._client_id = client_id
+            self.client_id = client_id
             self.service_name = service_name
             self.nats_config = {
                 "host": host,
@@ -181,20 +181,14 @@ def set_logger(
             )
         self.logger.logger = logging.getLogger(service_name)
 
-    def start(self, from_another_thread=False):
-        if self.http_server and from_another_thread is False:
-            self._start()
-        else:
-            start_thread(self._start())
-
-    def _start(self):
+    def start(self):
         if self.logger_required:
             self.set_logger(
                 self.service_name,
                 self.app_root_path,
                 self.logger_files_path,
                 self.logger_in_separate_process,
-                self._client_id,
+                self.client_id,
             )
 
         try:
@@ -214,13 +208,18 @@ def _start(self):
             raise InitializingEventManagerError(error)
 
         self.nats_config["listen_subjects_callbacks"] = subjects_and_callbacks
-
         NATSClient.__init__(self, **self.nats_config)
         if self.app_strategy == "asyncio":
-            self.connector.publish_sync(f'panini_events.{self.service_name}.{self._client_id}.started', {})
-        
+            asyncio.ensure_future(
+                self.connector.client.publish(
+                    f"panini_events.{self.service_name}.{self.client_id}.started",
+                    b"{}",
+                )
+            )
+
         self.tasks = self.tasks + self.TASKS
         self.interval_tasks = self.INTERVAL_TASKS
+
         self._start_tasks()
 
     def _start_tasks(self):
diff --git a/panini/emulator_client.py b/panini/emulator_client.py
new file mode 100644
index 0000000..2102cb2
--- /dev/null
+++ b/panini/emulator_client.py
@@ -0,0 +1,171 @@
+import asyncio
+import json
+import threading
+import time
+
+from nats.aio.client import Client as NATS
+
+
+
+def _dict_to_bytes(message: dict) -> bytes:
+    return json.dumps(message).encode('utf-8')
+
+
+def _bytes_to_dict(payload: bytes) -> dict:
+    return json.loads(payload)
+
+
+class EmulatorClient(threading.Thread):
+
+    def __init__(
+            self,
+            filepath: str,
+            prefix: str,
+            app_name: str,
+            emulate_timeout: bool = True,
+            compare_output: bool = False,
+            max_timeout_after_start: float = 20.0,
+    ):
+        threading.Thread.__init__(self)
+
+        self._name = "emulator_client" + prefix
+        self._filepath = filepath
+
+        self._publish_queue = []
+        self._listen_queues = {}
+
+        self._emulate_timeout = emulate_timeout
+        self._compare_output = compare_output
+        self._max_timeout_after_start = max_timeout_after_start
+        self._client = NATS()
+
+        self._app_name = app_name
+
+        self._prefix = prefix
+        self._reply_to_suffix = "reply"
+        self._subscriptions = []
+
+        self._is_app_started = False
+        self._is_emulator_ready = False
+
+        self._load()
+
+    def _load(self):
+        with open(self._filepath) as file:
+            lines = file.readlines()
+
+        for line in lines:
+            event = json.loads(line)
+            event_type = event["event_type"]
+            subject = self._prefix + "." + event["subject"]
+
+            if event_type.startswith("listen"):
+                self._publish_queue.append(event)
+
+            elif event_type.startswith("send"):
+                if subject not in self._listen_queues:
+                    self._listen_queues[subject] = []
+
+                self._listen_queues[subject].append(event)
+
+    async def _mock_request(self, message):
+        subject = message.subject
+        reply_to = message.reply
+        body = _bytes_to_dict(message.data)
+        event = self._listen_queues[subject].pop(0)
+        if self._compare_output:
+            assert body == event["message"], f"'{body}' vs '{event['message']}'"
+
+        if event["event_type"].endswith("request"):
+            await self._client.publish(reply_to, _dict_to_bytes(event["response"]))
+
+    def listen(self, subject: str):
+        def decorator(func):
+            assert isinstance(subject, str)
+
+            def wrapper(incoming_response):
+                wrapper_response = func(topic=incoming_response.subject,
+                                        message=_bytes_to_dict(incoming_response.data))
+                if wrapper_response is not None and incoming_response.reply != "":
+                    self._client.publish(incoming_response.reply, wrapper_response)
+
+            self._subscriptions.append([self._prefix + "." + subject, wrapper])
+
+            return wrapper
+
+        return decorator
+
+    def run(self):
+        loop = asyncio.new_event_loop()
+        loop.run_until_complete(self._run())
+
+    async def _on_app_started(self, message):
+        self._is_app_started = True
+
+    async def _run(self):
+        await self._client.connect()
+
+        for subject in self._listen_queues:
+            await self._client.subscribe(subject, cb=self._mock_request)
+
+        for subject, callback in self._subscriptions:
+            await self._client.subscribe(subject, cb=callback)
+
+        subject = f"{self._prefix}.panini_events.{self._app_name}.*.started"
+        await self._client.subscribe(subject, cb=self._on_app_started)
+
+        self._is_emulator_ready = True
+        await self._wait_for_app_to_start()
+        # publish all the events
+        await self._run_publish()
+
+        await self._wait_after()
+
+    async def _run_publish(self):
+        last_timestamp = None
+        while len(self._publish_queue) > 0:
+            event = self._publish_queue.pop(0)
+
+            event_type = event["event_type"]
+            subject = self._prefix + "." + event["subject"]
+            timestamp = event["timestamp"]
+            message = _dict_to_bytes(event["message"])
+
+            if event_type.endswith("publish"):
+                await self._client.publish(subject, message)
+
+            elif event_type.endswith("request"):
+                response = await self._client.request(subject, message, timeout=5)
+
+                if self._reply_to_suffix:
+                    await self._client.publish(f"{subject}.{self._reply_to_suffix}",
+                                               _dict_to_bytes(event["response"]))
+
+                if self._compare_output:
+                    assert json.loads(response.data) == event["response"]
+
+            if self._emulate_timeout:
+                if last_timestamp:
+                    wait = timestamp - last_timestamp
+                    time.sleep(wait)
+
+                last_timestamp = timestamp
+
+    def wait_for_readiness(self):
+        while not self._is_emulator_ready:
+            time.sleep(0.1)
+
+    async def _wait_for_app_to_start(self):
+        while not self._is_app_started:
+            await asyncio.sleep(0.1)
+
+    async def _wait_after(self):
+        start = time.time()
+        while True:
+            if sum(len(queue) for queue in self._listen_queues.values()) == 0:
+                break
+
+            if self._max_timeout_after_start and time.time() - start > self._max_timeout_after_start:
+                break
+
+            await asyncio.sleep(0.1)
diff --git a/panini/exceptions.py b/panini/exceptions.py
index 46fc58e..93556ec 100644
--- a/panini/exceptions.py
+++ b/panini/exceptions.py
@@ -95,3 +95,7 @@ class DecodeError(BaseError):
 
 class DataTypeError(BaseError):
     pass
+
+
+class TestClientError(BaseError):
+    pass
diff --git a/panini/middleware/reader_emulator_middleware.py b/panini/middleware/reader_emulator_middleware.py
new file mode 100644
index 0000000..c267bcd
--- /dev/null
+++ b/panini/middleware/reader_emulator_middleware.py
@@ -0,0 +1,48 @@
+import uuid
+
+from panini.app import get_app
+import copy
+
+from panini.emulator_client import EmulatorClient
+from panini.middleware import Middleware
+from panini.nats_client.nats_client_interface import Msg
+from panini.utils.helper import start_thread
+
+
+class ReaderEmulatorMiddleware(Middleware):
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(args, kwargs)
+
+        self._prefix = kwargs.get("prefix", str(uuid.uuid4()))
+
+        app = get_app()
+        assert app is not None
+        app.logger.warning('Emulator is experimental feature!')
+        subjects = copy.copy(list(app.SUBSCRIPTIONS.keys()))
+        for subject in subjects:
+            subject_with_prefix = self._prefix + "." + subject
+            app.SUBSCRIPTIONS[subject_with_prefix] = app.SUBSCRIPTIONS.pop(subject)
+
+        self._run_emulator = kwargs.get("run_emulator", True)
+
+        if self._run_emulator:
+            emulator = EmulatorClient(
+                prefix=self._prefix,
+                filepath=kwargs.get("filename"),
+                emulate_timeout=kwargs.get("emulate_timeout", True),
+                compare_output=kwargs.get("compare_output", False),
+                app_name=app.service_name
+            )
+            start_thread(emulator.start)
+            emulator.wait_for_readiness()
+
+    async def listen_any(self, message: Msg, callback):
+        message.subject = message.subject[len(self._prefix) + 1:]
+        response = await callback(message)
+        return response
+
+    async def send_any(self, subject: str, message, send_func, *args, **kwargs):
+        subject = self._prefix + "." + subject
+        response = await send_func(subject, message, *args, **kwargs)
+        return response
diff --git a/panini/middleware/writer_emulator_middleware.py b/panini/middleware/writer_emulator_middleware.py
new file mode 100644
index 0000000..195c4f3
--- /dev/null
+++ b/panini/middleware/writer_emulator_middleware.py
@@ -0,0 +1,130 @@
+import json
+import os
+import threading
+import time
+import collections
+from datetime import datetime
+from typing import Any
+
+from panini import middleware
+from panini.app import get_app
+
+
+class _Writer(threading.Thread):
+
+    def __init__(self, filename: str):
+        threading.Thread.__init__(self)
+
+        self._array = collections.deque()
+        self._lock = threading.Lock()
+
+        self._filename = filename
+
+    def add(self, event: dict):
+        with self._lock:
+            self._array.append(event)
+
+    def run(self):
+
+        """
+            This is a main method of _Writer.
+            It checks the array in order to write it to the file with events
+
+            It keeps the file opened if there are any messages, to do not spend time for reopen every time
+
+            Also, it has gradually increasing sleep timer
+
+        """
+
+        file = None
+        min_sleep_time = 0.1
+        max_sleep_time = 1
+        sleep_time = min_sleep_time
+        while True:
+            try:
+                if len(self._array) == 0:
+                    # close file only when array is empty
+                    if file:
+                        file.close()
+                        file = None
+
+                    # gradually increase sleep time, in order to handle appeared massages faster
+                    time.sleep(sleep_time)
+                    sleep_time = min(max_sleep_time, sleep_time * 2)
+                else:
+                    if not file:
+                        file = open(self._filename, 'a')
+
+                    line = self._array.popleft()
+                    file.write(json.dumps(line) + os.linesep)
+                    sleep_time = min_sleep_time
+            except Exception as ex:
+                print(f"exception: {ex}")
+
+
+class WriterEmulatorMiddleware(middleware.Middleware):
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(args, kwargs)
+
+        app = get_app()
+        app.logger.warning('Emulator is experimental feature!')
+        folder = kwargs.get("folder")
+        if not os.path.isdir(folder):
+            os.makedirs(folder)
+
+        filename = f"{folder}/events.{app.service_name}.{datetime.now().strftime('%Y-%m-%d-%H:%M:%S')}.jsonl"
+        self._prefix = kwargs.get("prefix", "emulator")
+        self._writer = _Writer(filename)
+        self._writer.start()
+
+    async def send_publish(self, subject: str, message: Any, publish_func, **kwargs):
+
+        self._writer.add({
+            "event_type": "send_publish",
+            "subject": subject,
+            "message": message,
+            "timestamp": time.time()
+        })
+
+        await publish_func(subject, message, **kwargs)
+
+    async def send_request(self, subject: str, message: Any, request_func, **kwargs):
+        response = await request_func(subject, message, **kwargs)
+
+        self._writer.add({
+            "event_type": "send_request",
+            "subject": subject,
+            "message": message,
+            "response": response,
+            "timestamp": time.time()
+        })
+
+        return response
+
+    async def listen_publish(self, msg, callback):
+
+        self._writer.add({
+            "event_type": "listen_publish",
+            "subject": msg.subject,
+            "message": msg.data,
+            "context": msg.context,
+            "timestamp": time.time()
+        })
+
+        await callback(msg)
+
+    async def listen_request(self, msg, callback):
+
+        response = await callback(msg)
+
+        self._writer.add({
+            "event_type": "listen_request",
+            "subject": msg.subject,
+            "message": msg.data,
+            "context": msg.context,
+            "response": response,
+            "timestamp": time.time()
+        })
+
+        return response
diff --git a/panini/nats_client/nats_client.py b/panini/nats_client/nats_client.py
index e715439..a948ae8 100644
--- a/panini/nats_client/nats_client.py
+++ b/panini/nats_client/nats_client.py
@@ -90,15 +90,9 @@ def inject_send_middlewares(self):
         self.connector.publish = _MiddlewareManager._wrap_function_by_middleware(
             self.connector.publish, "publish"
         )
-        self.connector.publish_sync = _MiddlewareManager._wrap_function_by_middleware(
-            self.connector.publish_sync, "publish"
-        )
         self.connector.request = _MiddlewareManager._wrap_function_by_middleware(
             self.connector.request, "request"
         )
-        self.connector.request_sync = _MiddlewareManager._wrap_function_by_middleware(
-            self.connector.request_sync, "request"
-        )
 
     def check_connection(self):
         self.connector.check_connection()
diff --git a/setup.py b/setup.py
index 6049dc2..7118318 100644
--- a/setup.py
+++ b/setup.py
@@ -2,8 +2,8 @@
 from setuptools import setup, find_packages
 
 long_description = """
-Panini is a modern framework for quick development of streaming microservices. Our goal is to create fastapi/aiohttp/flask-like solution but for NATS streaming.
- 
+Panini is a modern framework for quick development of streaming microservices. 
+Our goal is to create fastapi/aiohttp/flask-like solution but for NATS streaming.
 The framework allows you to work with NATS features and some additional logic using a simple interface:
 
 - publish messages to subject
@@ -19,12 +19,12 @@
 
 setup(
     name="panini",
-    version="0.2.1",
+    version="0.2.2",
     description="A python messaging framework for microservices based on NATS",
     long_description=long_description,
     long_description_content_type="text/x-rst",
     url="https://github.com/lwinterface/panini",
-    author="Op Return SA, developers: Andrew Volotskov, Danylo Tiutiushkin",
+    author="Op Return SA, developers: Andrii Volotskov, Danylo Tiutiushkin, Oleksii Volotskov",
     author_email="example@example.com",
     python_requires=">=3.8.2",
     license="MIT",


Test patch:
diff --git a/examples/emulator/tested_listener_app.py b/examples/emulator/tested_listener_app.py
new file mode 100644
index 0000000..600c1e0
--- /dev/null
+++ b/examples/emulator/tested_listener_app.py
@@ -0,0 +1,20 @@
+from panini.emulator_client import EmulatorClient
+
+filename = "resources/events.listener.2021-03-19-16:21:13.jsonl"
+
+client = EmulatorClient(
+    filepath=filename,
+    prefix='prefix',
+    app_name='listener'
+)
+
+
+# check response from target app
+@client.listen("listener.store.request.reply")
+def request_reply(topic, message):
+    print(f"request reply, {topic} {message}")
+
+
+
+if __name__ == "__main__":
+    client.run()
diff --git a/examples/emulator/tested_publisher_app.py b/examples/emulator/tested_publisher_app.py
new file mode 100644
index 0000000..ba16f08
--- /dev/null
+++ b/examples/emulator/tested_publisher_app.py
@@ -0,0 +1,23 @@
+from panini.emulator_client import EmulatorClient
+
+filename = "resources/events.publisher.2021-03-19-16:21:30.jsonl"
+
+client = EmulatorClient(
+    filepath=filename,
+    prefix='prefix',
+    app_name='publisher',
+)
+
+
+@client.listen("listener.store.listen")
+async def listen(topic, message):
+    print(topic, message)
+
+
+@client.listen("listener.store.request")
+async def response(topic, message):
+    print(topic, message)
+
+
+if __name__ == "__main__":
+    client.run()
diff --git a/panini/test_client.py b/panini/test_client.py
index 9f92c5a..e349a5e 100644
--- a/panini/test_client.py
+++ b/panini/test_client.py
@@ -10,6 +10,7 @@
 import websocket
 from pynats import NATSClient, NATSMessage
 
+from .exceptions import TestClientError
 from .utils.helper import start_process
 from .nats_client.nats_client_interface import Msg
 
@@ -82,11 +83,13 @@ class TestClient:
     def __init__(
         self,
         run_panini: typing.Callable = lambda *args, **kwargs: None,
+        panini_service_name: str = "*",
+        panini_client_id: str = "*",
         use_web_server: bool = False,
         use_web_socket: bool = False,
         base_web_server_url: str = "http://127.0.0.1:8080",
         base_nats_url: str = "nats://127.0.0.1:4222",
-        socket_timeout: int = 2,
+        socket_timeout: int = 5,
         name: str = "__".join(
             [
                 "test_client",
@@ -96,6 +99,11 @@ def __init__(
         ),
     ):
         self.run_panini = run_panini
+        self.panini_service_name = panini_service_name
+        self.panini_client_id = panini_client_id
+        self.base_web_server_url = base_web_server_url
+        self.use_web_server = use_web_server
+        self.use_web_socket = use_web_socket
         self.nats_client = NATSClient(
             url=base_nats_url,
             name=name,
@@ -129,19 +137,33 @@ def wrap_run_panini(run_panini):
         except Exception as e:
             test_logger.exception(f"Run panini error: {e}")
 
-    def start(
-        self, is_sync: bool = False, sleep_time: float = None, is_daemon: bool = None
-    ):
+    def start(self, is_sync: bool = False, is_daemon: bool = None):
         if is_daemon is None:
             is_daemon = False if is_sync else True
 
+        @self.listen(
+            f"panini_events.{self.panini_service_name}.{self.panini_client_id}.started"
+        )
+        def panini_started(msg):
+            pass
+
         self.panini_process = start_process(
             self.wrap_run_panini, args=(self.run_panini,), daemon=is_daemon
         )
-        if sleep_time is None:
-            time.sleep(6) if is_sync else time.sleep(1)
+
+        if is_sync:
+            time.sleep(5)
         else:
-            time.sleep(sleep_time)
+            try:
+                self.wait(1)  # wait for panini to start
+            except OSError:
+                raise TestClientError(
+                    "TestClient was waiting panini to start, but panini does not started"
+                )
+
+        if self.use_web_server:
+            pass  # TODO: understand, why don't we need to wait for web_server
+
         return self
 
     def stop(self):
@@ -200,7 +222,13 @@ def wrapper(incoming_response):
                     self.publish(
                         subject=incoming_response.reply, message=wrapper_response
                     )
-                    self.wait(count=1)
+                    try:
+                        self.wait(count=1)
+                    except OSError:
+                        raise TestClientError(
+                            f"TestClient listen subject: {incoming_response.subject},"
+                            f"Response was sent, but it was not correctly handled"
+                        )
 
             self.subscribe(subject, wrapper)
 
diff --git a/tests/emulator/test_event_storage_middleware.py b/tests/emulator/test_event_storage_middleware.py
new file mode 100644
index 0000000..64caa83
--- /dev/null
+++ b/tests/emulator/test_event_storage_middleware.py
@@ -0,0 +1,107 @@
+import time
+
+import pytest
+
+from panini.middleware.writer_emulator_middleware import WriterEmulatorMiddleware
+from panini.test_client import TestClient
+from panini import app as panini_app
+from tests.helper import Global
+
+
+def run_panini():
+    app = panini_app.App(
+        service_name="test_middleware",
+        host="127.0.0.1",
+        port=4222,
+        logger_in_separate_process=False
+    )
+
+    @app.listen("test_middleware.publish")
+    async def publish(msg):
+        try:
+            await app.publish(
+                subject="test_middleware.publish.response",
+                message={"data": msg.data["data"] + 1},
+            )
+        except Exception:
+            app.logger.exception("test_middleware.publish")
+
+    @app.listen("test_middleware.request")
+    async def request(msg):
+        try:
+            response = await app.request(
+                subject="test_middleware.request.helper",
+                message={"data": msg.data["data"]},
+            )
+            return {"data": response["data"] + 2}
+        except Exception:
+            app.logger.exception("test_middleware.request")
+
+    @app.listen("test_middleware.request.helper")
+    async def helper(msg):
+        try:
+            return {"data": msg.data["data"] + 2}
+        except Exception:
+            app.logger.exception("test_middleware.request.helper")
+
+    @app.listen("test_middleware.listen.publish")
+    async def request(msg):
+        try:
+            await app.publish(
+                subject="test_middleware.listen.publish.response",
+                message={"data": msg.data["data"] + 3},
+            )
+        except Exception:
+            app.logger.exception("test_middleware.listen.publish")
+
+    @app.listen("test_middleware.listen.request")
+    async def request(msg):
+        try:
+            return {"data": msg.data["data"] + 4}
+        except Exception:
+            app.logger.exception("test_middleware.listen.request")
+
+    folder = "../resources"
+    app.add_middleware(WriterEmulatorMiddleware, folder=folder)
+    app.start()
+
+
+global_object = Global()
+
+
+@pytest.fixture(scope="session")
+def client():
+    client = TestClient(run_panini)
+
+    @client.listen("test_middleware.publish.response")
+    def publish_listener(msg):
+        pass
+
+
+    @client.listen("test_middleware.listen.publish.response")
+    def listen_publish_listener(msg):
+        pass
+
+    client.start()
+    return client
+
+
+def test_send_publish_middleware(client):
+    client.publish("test_middleware.publish", {"data": 1})
+    client.wait(1)
+
+def test_send_request_middleware(client):
+    response = client.request("test_middleware.request", {"data": 2})
+
+
+def test_listen_publish_middleware(client):
+    client.publish("test_middleware.listen.publish", {"data": 3})
+    client.wait(1)
+
+
+def test_listen_request_middleware(client):
+    response = client.request("test_middleware.listen.request", {"data": 4})
+
+
+def test_sleep(client):
+    time.sleep(2)
\ No newline at end of file
diff --git a/tests/run_test.sh b/tests/run_test.sh
old mode 100644
new mode 100755
diff --git a/tests/test_diff_datatypes.py b/tests/test_diff_datatypes.py
index 485f5f5..2b1b913 100644
--- a/tests/test_diff_datatypes.py
+++ b/tests/test_diff_datatypes.py
@@ -102,13 +102,15 @@ async def request_bytes(msg):
 
 @pytest.fixture(scope="module")
 def client():
-    client = TestClient(run_panini).start()
+    client = TestClient(run_panini)
 
     @client.listen("test_diff_datatypes.publish.listener")
     def dict_listener(msg):
         global_object.public_variable = msg.data["type"]
 
+    client.start()
     yield client
+    client.stop()
 
 
 def test_listen_dict(client):
diff --git a/tests/test_logs.py b/tests/test_logs.py
index c432ee5..50b6ae7 100644
--- a/tests/test_logs.py
+++ b/tests/test_logs.py
@@ -40,8 +40,9 @@ async def composite_subject_for_requests(msg):
 @pytest.fixture(scope="module")
 def client():
     client = TestClient(run_panini)
-    client.start(sleep_time=2)
+    client.start()
     yield client
+    client.stop()
 
 
 def test_simple_log(client):
diff --git a/tests/test_logs_in_separate_process.py b/tests/test_logs_in_separate_process.py
index 9f15070..9803db8 100644
--- a/tests/test_logs_in_separate_process.py
+++ b/tests/test_logs_in_separate_process.py
@@ -46,7 +46,7 @@ async def kill_logs(msg):
 @pytest.fixture(scope="module")
 def client():
     client = TestClient(run_panini)
-    client.start(sleep_time=2, is_daemon=False)
+    client.start(is_daemon=False)
     yield client
     client.stop()
 
@@ -56,7 +56,7 @@ def test_simple_log(client):
     assert response["success"] is True
 
     # wait for log being written
-    time.sleep(0.1)
+    time.sleep(1)
     with open(
         os.path.join(testing_logs_directory_path, "test_logs_in_separate_process.log"),
         "r",
diff --git a/tests/test_middleware.py b/tests/test_middleware.py
index 42a5c0d..b660986 100644
--- a/tests/test_middleware.py
+++ b/tests/test_middleware.py
@@ -7,12 +7,12 @@
 
 
 class AddMiddleware(Middleware):
-    async def send_publish(self, subject: str, message, publish_func, **kwargs):
+    async def send_publish(self, subject: str, message, publish_func, *args, **kwargs):
         message["data"] += 1
-        await publish_func(subject, message, **kwargs)
+        await publish_func(subject, message, *args, **kwargs)
 
-    async def send_request(self, subject: str, message, request_func, **kwargs):
-        response = await request_func(subject, message, **kwargs)
+    async def send_request(self, subject: str, message, request_func, *args, **kwargs):
+        response = await request_func(subject, message, *args, **kwargs)
         response["data"] += 2
         return response
 
diff --git a/tests/test_parameter_allocation_queue_group.py b/tests/test_parameter_allocation_queue_group.py
deleted file mode 100755
index 1df395b..0000000
--- a/tests/test_parameter_allocation_queue_group.py
+++ /dev/null
@@ -1,70 +0,0 @@
-import time
-import pytest
-
-from panini.test_client import TestClient, get_logger_files_path
-from panini import app as panini_app
-
-from panini.utils.helper import start_process
-
-
-def run_panini1():
-    app = panini_app.App(
-        service_name="test_allocation_queue_group1",
-        host="127.0.0.1",
-        port=4222,
-        allocation_queue_group="group1",
-        app_strategy="asyncio",
-        logger_in_separate_process=False,
-        logger_files_path=get_logger_files_path(),
-    )
-
-    @app.listen("test_parameter_allocation_queue_group.foo")
-    async def foo(msg):
-        return {"data": 1}
-
-    app.start()
-
-
-def run_panini2():
-    app = panini_app.App(
-        service_name="test_allocation_queue_group2",
-        host="127.0.0.1",
-        port=4222,
-        allocation_queue_group="group1",
-        app_strategy="asyncio",
-        logger_in_separate_process=False,
-        logger_files_path=get_logger_files_path(),
-    )
-
-    @app.listen("test_parameter_allocation_queue_group.foo")
-    async def foo(msg):
-        return {"data": 2}
-
-    app.start()
-
-
-@pytest.fixture(scope="module")
-def client():
-    client = TestClient()
-    # if you want to run more that 1 panini app in testing, please use start_process function for each app
-    start_process(run_panini1)
-    start_process(run_panini2)
-    # wait for panini apps to setup
-    time.sleep(2)
-    yield client
-    client.stop()
-
-
-def test_listen_subject_only_if_include_one_request(client):
-    response = client.request("test_parameter_allocation_queue_group.foo", {})
-    assert response["data"] in (1, 2)
-
-
-def test_listen_subject_only_if_include_multiple_requests(client):
-    """Tests that some requests are handled by first panini app and some by second"""
-    results = set(
-        client.request("test_parameter_allocation_queue_group.foo", {})["data"]
-        for _ in range(10)
-    )
-    assert 1 in results
-    assert 2 in results
diff --git a/tests/test_web_server.py b/tests/test_web_server.py
index dcc27e2..927a1fe 100755
--- a/tests/test_web_server.py
+++ b/tests/test_web_server.py
@@ -41,7 +41,7 @@ def client():
         use_web_server=True,
         base_web_server_url="http://127.0.0.1:8084",
     )
-    client.start(sleep_time=3)
+    client.start()
     yield client
     client.stop()
 
diff --git a/tests/test_wss.py b/tests/test_wss.py
index a8d07a9..29c8b8b 100644
--- a/tests/test_wss.py
+++ b/tests/test_wss.py
@@ -52,7 +52,7 @@ async def incoming_messages_callback(subscriber, msg):
 def client():
     # provide parameter for using web_socket - use_web_socket;
     client = TestClient(run_panini=run_panini, use_web_socket=True)
-    client.start(sleep_time=3)
+    client.start()
     yield client
     client.stop()