Skip to content
Permalink
Browse files

fix(stream-client): request queue size is limited by 1000

  • Loading branch information...
numb3r3 committed Sep 19, 2019
1 parent 4f38944 commit 05db02f71e8b6a38249ed6a08417e2cf3e73df14
Showing with 16 additions and 7 deletions.
  1. +16 −7 gnes/client/stream.py
@@ -44,27 +44,36 @@ class StreamingClient(GrpcClient):
def __init__(self, args):
super().__init__(args)

self._request_queue = queue.Queue()
self._request_queue = queue.Queue(maxsize=1000)
self._is_streaming = threading.Event()

self._dispatch_thread = threading.Thread(target=self._start)
self._dispatch_thread.setDaemon(1)
self._dispatch_thread.start()
self._dispatch_thread.setDaemon(True)

def send_request(self, request):
self._request_queue.put(request)
self._request_queue.put(request, block=True)

# create a new streaming call
if not self._is_streaming.is_set():
self._dispatch_thread.start()

def _start(self):
self._is_streaming.set()
response_stream = self.stream_call(self._request_generator())
self.stream_call(self._request_generator())
self._is_streaming.clear()

def _request_generator(self):
while self._is_streaming.is_set():
try:
request = self._request_queue.get(block=True, timeout=1.0)
request = self._request_queue.get(block=True, timeout=5.0)
if request is None:
break
yield request
except queue.Empty:
pass
continue
except Exception as e:
print('exception: %s' % str(e))
break

@handler.register(NotImplementedError)
def _handler_default(self, resp: 'gnes_pb2.Response'):

0 comments on commit 05db02f

Please sign in to comment.
You can’t perform that action at this time.