Fix session pulses never getting sent (#143)

## What is the goal of this PR?

The existing implementation of session renewal was not working, at all. Sessions would simply time out after 10 seconds. We fixed it.

## What are the changes implemented in this PR?

Fix session pulses never getting sent

Author: Alex Walker

grakn/client.py

@@ -18,8 +18,6 @@
 #
 
 import grpc
-import sched
-import time
 
 from grakn.options import GraknOptions
 from grakn.rpc.database_manager import DatabaseManager as _DatabaseManager
@@ -36,7 +34,6 @@ class GraknClient(object):
     def __init__(self, address=DEFAULT_URI):
         self._channel = grpc.insecure_channel(address)
         self._databases = _DatabaseManager(self._channel)
-        self._scheduler = sched.scheduler(time.time, time.sleep)
 
     def session(self, database: str, session_type: SessionType, options=GraknOptions()):
         return _Session(self, database, session_type, options)

grakn/rpc/session.py

@@ -16,6 +16,9 @@
 # specific language governing permissions and limitations
 # under the License.
 #
+import sched
+import time
+from threading import Thread
 
 from graknprotocol.protobuf.grakn_pb2_grpc import GraknStub
 import graknprotocol.protobuf.session_pb2 as session_proto
@@ -33,9 +36,11 @@ class SessionType(enum.Enum):
 
 class Session(object):
 
+    _PULSE_FREQUENCY_SECONDS = 5
+
     def __init__(self, client, database: str, session_type: SessionType, options=GraknOptions()):
         self._channel = client._channel
-        self._scheduler = client._scheduler
+        self._scheduler = sched.scheduler(time.time, time.sleep)
         self._database = database
         self._session_type = session_type
         self._grpc_stub = GraknStub(self._channel)
@@ -47,7 +52,10 @@ def __init__(self, client, database: str, session_type: SessionType, options=Gra
 
         self._session_id = self._grpc_stub.session_open(open_req).session_id
         self._is_open = True
-        self._pulse = self._scheduler.enter(5, 1, self._transmit_pulse, ())
+        self._scheduler.enter(delay=self._PULSE_FREQUENCY_SECONDS, priority=1, action=self._transmit_pulse, argument=())
+        # TODO: This thread blocks the process from closing. We should try cancelling the scheduled task when the
+        #       session closes. If that doesn't work, we need some other way of getting the thread to exit.
+        Thread(target=self._scheduler.run).start()
 
     def transaction(self, transaction_type: TransactionType, options=GraknOptions()):
         return Transaction(self._channel, self._session_id, transaction_type, options)
@@ -70,9 +78,10 @@ def _transmit_pulse(self):
             return
         pulse_req = session_proto.Session.Pulse.Req()
         pulse_req.session_id = self._session_id
-        is_alive = self._grpc_stub.session_pulse(pulse_req).is_alive
-        if is_alive:
-            self._pulse = self._scheduler.enter(5, 1, self._transmit_pulse, ())
+        res = self._grpc_stub.session_pulse(pulse_req)
+        if res.alive:
+            self._scheduler.enter(delay=self._PULSE_FREQUENCY_SECONDS, priority=1, action=self._transmit_pulse, argument=())
+            Thread(target=self._scheduler.run).start()
 
     def __enter__(self):
         return self