diff --git a/misc/eddn.py b/misc/eddn.py index 8b21732f..2ddaacc4 100644 --- a/misc/eddn.py +++ b/misc/eddn.py @@ -142,10 +142,14 @@ def connect(self): Start a connection """ # tear up the new connection first + if self.subscriber: + self.subscriber.close() + del self.subscriber self.subscriber = newsub = self.zmqContext.socket(zmq.SUB) newsub.setsockopt(zmq.SUBSCRIBE, b"") newsub.connect(self.uri) self.lastRecv = time.time() + self.lastJsData = None def disconnect(self): @@ -196,6 +200,8 @@ def get_batch(self, onerror=None): built-in auto-reconnection if there is nothing from the firehose for a period of time. + As json data is decoded, it is stored in self.lastJsData. + Parameters: onerror None or a function/lambda that takes an error @@ -233,6 +239,7 @@ def get_batch(self, onerror=None): # we reach the burst limit or we get EAGAIN. bursts = 0 for _ in range(self.burstLimit): + self.lastJsData = None try: zdata = sub.recv(flags=zmq.NOBLOCK, copy=False) stats['recvs'] += 1 @@ -259,6 +266,8 @@ def get_batch(self, onerror=None): onerror("json.loads: %s: %s"%(type(e), e)) continue + self.lastJsData = jsdata + try: schema = data["$schemaRef"] except KeyError: