Fix: queue.get(block=False) can raise Empty (#99)

Author: tom-pytel

skywalking/agent/protocol/grpc.py

@@ -18,7 +18,7 @@
 import logging
 from skywalking.loggings import logger
 import traceback
-from queue import Queue
+from queue import Queue, Empty
 
 import grpc
 
@@ -70,7 +70,10 @@ def on_error(self):
     def report(self, queue: Queue, block: bool = True):
         def generator():
             while True:
-                segment = queue.get(block=block)  # type: Segment
+                try:
+                    segment = queue.get(block=block)  # type: Segment
+                except Empty:
+                    return
 
                 logger.debug('reporting segment %s', segment)
 

skywalking/agent/protocol/http.py

@@ -16,7 +16,7 @@
 #
 
 from skywalking.loggings import logger
-from queue import Queue
+from queue import Queue, Empty
 
 from skywalking.agent import Protocol
 from skywalking.client.http import HttpServiceManagementClient, HttpTraceSegmentReportService
@@ -41,7 +41,10 @@ def connected(self):
     def report(self, queue: Queue, block: bool = True):
         def generator():
             while True:
-                segment = queue.get(block=block)  # type: Segment
+                try:
+                    segment = queue.get(block=block)  # type: Segment
+                except Empty:
+                    return
 
                 logger.debug('reporting segment %s', segment)
 

skywalking/agent/protocol/kafka.py

@@ -17,7 +17,7 @@
 
 import logging
 from skywalking.loggings import logger, getLogger
-from queue import Queue
+from queue import Queue, Empty
 
 from skywalking import config
 from skywalking.agent import Protocol
@@ -45,7 +45,10 @@ def heartbeat(self):
     def report(self, queue: Queue, block: bool = True):
         def generator():
             while True:
-                segment = queue.get(block=block)  # type: Segment
+                try:
+                    segment = queue.get(block=block)  # type: Segment
+                except Empty:
+                    return
 
                 logger.debug('reporting segment %s', segment)