stratum.py

   1 import json
   2 import socket
   3 import threading
   4 import time
   5 import Queue as queue
   6
   7 from processor import Session, Dispatcher, Shared
   8
   9 class TcpSession(Session):
  10
  11     def __init__(self, connection, address):
  12         self._connection = connection
  13         self.address = address
  14         Session.__init__(self)
  15
  16     def connection(self):
  17         if self.stopped():
  18             raise Exception("Session was stopped")
  19         else:
  20             return self._connection
  21
  22     def stop(self):
  23         self._connection.close()
  24         print "Terminating connection:", self.address[0]
  25         with self.lock:
  26             self._stopped = True
  27
  28     def send_response(self, response):
  29         raw_response = json.dumps(response)
  30         # Possible race condition here by having session
  31         # close connection?
  32         # I assume Python connections are thread safe interfaces
  33         try:
  34             connection = self.connection()
  35             connection.send(raw_response + "\n")
  36         except:
  37             self.stop()
  38
  39
  40
  41 class TcpClientRequestor(threading.Thread):
  42
  43     def __init__(self, shared, processor, session):
  44         self.shared = shared
  45         self.processor = processor
  46         self.message = ""
  47         self.session = session
  48         threading.Thread.__init__(self)
  49
  50     def run(self):
  51         while not self.shared.stopped():
  52             if not self.update():
  53                 break
  54
  55             while self.parse():
  56                 pass
  57
  58     def update(self):
  59         data = self.receive()
  60         if not data:
  61             # close_session
  62             self.session.stop()
  63             return False
  64
  65         self.message += data
  66         return True
  67
  68     def receive(self):
  69         try:
  70             return self.session.connection().recv(1024)
  71         except:
  72             return ''
  73
  74     def parse(self):
  75         raw_buffer = self.message.find('\n')
  76         if raw_buffer == -1:
  77             return False
  78
  79         raw_command = self.message[0:raw_buffer].strip()
  80         self.message = self.message[raw_buffer + 1:]
  81         if raw_command == 'quit':
  82             self.session.stop()
  83             return False
  84
  85         try:
  86             command = json.loads(raw_command)
  87         except:
  88             self.processor.push_response({"error": "bad JSON", "request": raw_command})
  89             return True
  90
  91         try:
  92             # Try to load vital fields, and return an error if
  93             # unsuccessful.
  94             message_id = command['id']
  95             method = command['method']
  96         except KeyError:
  97             # Return an error JSON in response.
  98             self.processor.push_response({"error": "syntax error", "request": raw_command})
  99         else:
 100             self.processor.push_request(self.session,command)
 101
 102         return True
 103
 104 class TcpServer(threading.Thread):
 105
 106     def __init__(self, shared, processor, host, port):
 107         self.shared = shared
 108         self.processor = processor
 109         threading.Thread.__init__(self)
 110         self.daemon = True
 111         self.host = host
 112         self.port = port
 113         self.lock = threading.Lock()
 114
 115     def run(self):
 116         print "TCP server started."
 117         sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 118         sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 119         sock.bind((self.host, self.port))
 120         sock.listen(1)
 121         while not self.shared.stopped():
 122             session = TcpSession(*sock.accept())
 123             client_req = TcpClientRequestor(self.shared, self.processor, session)
 124             client_req.start()
 125             self.processor.add_session(session)
 126             self.processor.collect_garbage()
 127
 128
 129
 130
 131 class Stratum:
 132
 133     def start(self, processor):
 134         shared = Shared()
 135         # Bind shared to processor since constructor is user defined
 136         processor.shared = shared
 137         processor.start()
 138         # Create various transports we need
 139         transports = TcpServer(shared, processor, "176.31.24.241", 50001),
 140         for server in transports:
 141             server.start()
 142         while not shared.stopped():
 143             if raw_input() == "quit":
 144                 shared.stop()
 145             time.sleep(1)
 146
 147 if __name__ == "__main__":
 148     processor = Processor()
 149     app = Stratum()
 150     app.start(processor)
 151