split internal classes into different files

Author: Time-Coder

Connector/AutoSendBuffer.py

@@ -0,0 +1,53 @@
+import threading
+import time
+
+
+class AutoSendBuffer:
+
+    def __init__(self, node, session_id, timeout=0.1, quiet=False):
+        self._node = node
+        self._session_id = session_id
+        self._buffer_lock = threading.Lock()
+        self._buffer = b''
+        self._quiet = quiet
+        self._timeout = timeout
+        self._last_send_time = time.time()
+        self._continue_send = True
+        self._sending_thread = threading.Thread(target=self._timeout_send)
+        self._sending_thread.start()
+
+    def __del__(self):
+        try:
+            self.stop()
+        except BaseException:
+            pass
+
+    def append(self, byte):
+        with self._buffer_lock:
+            self._buffer += byte
+
+        if byte == b"\n" or len(self._buffer) >= 512:
+            self.send()
+
+    def send(self):
+        with self._buffer_lock:
+            if len(self._buffer) != 0:
+                content = self._buffer.decode("utf-8")
+                if not self._quiet:
+                    print(content, end="", flush=True)
+                self._node._respond_ok(
+                    self._session_id, end=False, stdout=content
+                )
+                self._buffer = b''
+                self._last_send_time = time.time()
+
+    def stop(self):
+        self._continue_send = False
+        self._sending_thread.join()
+        self.send()
+
+    def _timeout_send(self):
+        while self._continue_send:
+            if time.time()-self._last_send_time >= self._timeout:
+                self.send()
+            time.sleep(1E-3)

Connector/Future.py

@@ -0,0 +1,84 @@
+import threading
+
+from .utils import eprint
+
+
+class Future:
+
+    def __init__(self, node, session_id):
+        self._node = node
+        self._session_id = session_id
+        self._cancelled = False
+        self._done = None
+        self._result = None
+        self._monitor_done_thread = None
+        self._done_callbacks = []
+
+    def cancel(self):
+        if self.done():
+            return False
+
+        self._node._stop_send = True
+        self._node._send_signal(self._session_id, cancel=True)
+        result_queues = self._node._result_queues_for_future
+        self._result = result_queues[self._session_id].get()
+        self._node._close_session(self._session_id)
+        self._cancelled = True
+        self._done = True
+        return True
+
+    def cancelled(self):
+        return self._cancelled
+
+    def done(self):
+        if self._done is not None:
+            return self._done
+        else:
+            result_queues = self._node._result_queues_for_future
+            return (len(result_queues[self._session_id]) > 0)
+
+    def result(self):
+        if self._result is None:
+            result_queues = self._node._result_queues_for_future
+            self._result = result_queues[self._session_id].get()
+            self._node._close_session(self._session_id)
+
+        self._done = True
+        if self._result["success"]:
+            return self._result["return_value"]
+        else:
+            eprint(self._result["traceback"])
+            raise self._result["exception"]
+
+    def exception(self):
+        if self._result is None:
+            result_queues = self._node._result_queues_for_future
+            self._result = result_queues[self._session_id].get()
+            self._node._close_session(self._session_id)
+
+        self._done = True
+        return self._result["exception"]
+
+    def _get_result(self):
+        if self._result is None:
+            result_queues = self._node._result_queues_for_future
+            self._result = result_queues[self._session_id].get()
+            self._node._close_session(self._session_id)
+
+        for func in self._done_callbacks:
+            func(self)
+
+    def add_done_callback(self, func):
+        self._done_callbacks.append(func)
+
+        if self._monitor_done_thread is None:
+            self._monitor_done_thread = threading.Thread(
+                target=self._get_result
+            )
+            self._monitor_done_thread.start()
+
+    def remove_done_callback(self, func):
+        try:
+            self._done_callbacks.remove(func)
+        except BaseException:
+            pass

Connector/KillableThread.py

@@ -0,0 +1,28 @@
+import threading
+import ctypes
+
+
+class KillableThread(threading.Thread):
+
+    def __init__(self, target, args=(), kwargs={}):
+        threading.Thread.__init__(
+            self, target=target,
+            args=args, kwargs=kwargs
+        )
+
+    def __get_id(self):
+        if hasattr(self, '_thread_id'):
+            return self._thread_id
+
+        for id, thread in threading._active.items():
+            if thread is self:
+                return id
+
+    def kill(self):
+        thread_id = self.__get_id()
+        res = ctypes.pythonapi.PyThreadState_SetAsyncExc(
+            thread_id, ctypes.py_object(SystemExit)
+        )
+        if res > 1:
+            ctypes.pythonapi.PyThreadState_SetAsyncExc(thread_id, 0)
+            print('Exception raise failure')

Connector/NodeBasicMethods.py

@@ -8,7 +8,9 @@
 
 from .CloseableQueue import CloseableQueue
 from .CloseablePipe import CloseablePipe
-from .NodeInternalClasses import Pipes, Queues, PipeDict, QueueDict, ServerPeer
+from .Pipes import PipeDict, Pipes
+from .Queues import QueueDict, Queues
+from .ServerPeer import ServerPeer
 
 complex_request = ["get_file", "put_file",
                    "get_folder", "put_folder",
@@ -104,9 +106,9 @@ def hold_on(self):
 def _close(self):
     if self._is_internal_closing or self.is_closed:
         return
-    
+
     self._is_internal_closing = True
-    
+
     self_address = self.address
     try:
         self._connection.close()
@@ -195,7 +197,7 @@ def _close(self):
 def close(self):
     if self._is_closing or self.is_closed:
         return
-    
+
     self._is_closing = True
 
     try:
@@ -409,7 +411,7 @@ def _recving_loop(self):
         except BaseException:
             self._close()
             return
-        
+
         self._recved_binary_queue.put(recved_binary)
 
 

Connector/NodeFileTransferMethods.py

@@ -3,7 +3,8 @@
 
 from .utils import file_size, md5, eprint
 from .Config import Config
-from .NodeInternalClasses import Future, Thread
+from .Future import Future
+from .KillableThread import KillableThread
 
 
 def _write_to_file(self, data, file_name):
@@ -148,7 +149,7 @@ def session():
 
         self._make_signal(session_id)
 
-    thread = Thread(target=session)
+    thread = KillableThread(target=session)
     thread.start()
     signal = self._recv_signal(session_id)
     if signal["cancel"] and thread.is_alive():
@@ -201,7 +202,7 @@ def session():
     if block:
         session()
     else:
-        thread = Thread(target=session)
+        thread = KillableThread(target=session)
         thread.start()
         return Future(self, session_id)
 
@@ -251,7 +252,7 @@ def session():
 
         self._make_signal(session_id)
 
-    thread = Thread(target=session)
+    thread = KillableThread(target=session)
     thread.start()
     signal = self._recv_signal(session_id)
     if signal["cancel"] and thread.is_alive():
@@ -444,7 +445,7 @@ def session():
 
         self._make_signal(session_id)
 
-    thread = Thread(target=session)
+    thread = KillableThread(target=session)
     thread.start()
     signal = self._recv_signal(session_id)
     if signal["cancel"] and thread.is_alive():
@@ -678,7 +679,7 @@ def session():
 
         self._make_signal(session_id)
 
-    thread = Thread(target=session)
+    thread = KillableThread(target=session)
     thread.start()
     signal = self._recv_signal(session_id)
     if signal["cancel"] and thread.is_alive():