From b69d6032b25345cc838b6af05fa3a6d757bb0e79 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Torma=20Krist=C3=B3f?= <tormakristof@tormakristof.eu>
Date: Sun, 25 Apr 2021 13:11:10 +0200
Subject: [PATCH] close at the end

---
 kafka/consumer.py  | 1 +
 rabbit/consumer.py | 2 +-
 zmq/consumer.py    | 2 +-
 3 files changed, 3 insertions(+), 2 deletions(-)

diff --git a/kafka/consumer.py b/kafka/consumer.py
index 5099855..e39d1d6 100644
--- a/kafka/consumer.py
+++ b/kafka/consumer.py
@@ -27,3 +27,4 @@ except KeyboardInterrupt:
             dt[dts] +=1
     for key in dt:
         print(key,",",dt[key])
+    consumer.close()
diff --git a/rabbit/consumer.py b/rabbit/consumer.py
index f9ca60f..cae5da1 100644
--- a/rabbit/consumer.py
+++ b/rabbit/consumer.py
@@ -23,7 +23,6 @@ try:
     channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True)
     channel.start_consuming()
 except KeyboardInterrupt:
-    connection.close()
     dt = {}
     for datetime in DATETIMES:
         dts = datetime.strftime("%m/%d/%Y, %H:%M:%S")
@@ -33,3 +32,4 @@ except KeyboardInterrupt:
             dt[dts] +=1
     for key in dt:
         print(key,",",dt[key])
+    connection.close()
diff --git a/zmq/consumer.py b/zmq/consumer.py
index 280d461..be6184e 100644
--- a/zmq/consumer.py
+++ b/zmq/consumer.py
@@ -17,7 +17,6 @@ try:
         m = socket.recv_string()
         DATETIMES.append(datetime.now())
 except KeyboardInterrupt:
-    socket.close()
     dt = {}
     for datetime in DATETIMES:
         dts = datetime.strftime("%m/%d/%Y, %H:%M:%S")
@@ -27,3 +26,4 @@ except KeyboardInterrupt:
             dt[dts] +=1
     for key in dt:
         print(key,",",dt[key])
+    socket.close()