Multi-threaded socket client for Scrolls

Question

This is a multi-threaded socket client written to talk to the Scrolls socket server. The idea is to send commands to the socket server and respond to messages received via callbacks. I've never done multi-threading before and would like the code reviewed for code correctness, best practices, and potential issues regarding how I'm handling threading.

GitHub

from Crypto.Cipher import PKCS1_v1_5
from Crypto.PublicKey import RSA
from base64 import b64encode
from threading import Thread
from Queue import Queue
import socket
import json
import time
class PingThread(Thread):
 def __init__(self, scrolls_client):
 self.scrolls_client = scrolls_client
 self.stopped = False
 Thread.__init__(self)
 def run(self):
 while not self.stopped:
 self.scrolls_client.send({'msg': 'Ping'})
 time.sleep(10)
class MessageThread(Thread):
 def __init__(self, scrolls_client):
 self.scrolls_client = scrolls_client
 self.stopped = False
 Thread.__init__(self)
 def run(self):
 while not self.stopped:
 # grab a message from queue
 message = self.scrolls_client.queue.get()
 # make a copy of the current subscribers to keep this thread-safe
 current_subscribers = dict(self.scrolls_client.subscribers)
 # send message to subscribers
 for subscriber_key, subscriber_callback in current_subscribers.iteritems():
 # msg or op should match what we asked for
 if 'msg' in message and message['msg'] == subscriber_key:
 subscriber_callback(message)
 elif 'op' in message and message['op'] == subscriber_key:
 subscriber_callback(message)
 # signals to queue job is done
 self.scrolls_client.queue.task_done()
class ReceiveThread(Thread):
 def __init__(self, scrolls_client):
 self.scrolls_client = scrolls_client
 self.stopped = False
 Thread.__init__(self)
 def run(self):
 while not self.stopped:
 self.scrolls_client.receive()
class ScrollsSocketClient(object):
 """
 A Python client for the Scrolls socket server.
 Usage:
 YOUR_SCROLLS_EMAIL = '[email protected]'
 YOUR_SCROLLS_PASSWORD = 'password'
 scrolls = ScrollsApi(YOUR_SCROLLS_EMAIL, YOUR_SCROLLS_PASSWORD)
 """
 queue = Queue()
 subscribers = {}
 _socket_recv = 8192
 _scrolls_host = '54.208.22.193'
 _scrolls_port = 8081
 _scrolls_publickey = """-----BEGIN PUBLIC KEY-----
MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQCYUK5tWE8Yb564e5VBs05uqh38
mLSRF76iHY4IVHtpXT3FiI6SWoVDyOAiAAe/IJwzUmjCp8V4nmNX26nQuHR4iK/c
U9G7XhpBLfmQx0Esx5tJbYM0GR9Ww4XeXj3xZZBL39MciohrFurBENTFtrlu0EtM
3T8DbLpZaJeXTle7VwIDAQAB
-----END PUBLIC KEY-----"""
 def __init__(self, email, password):
 self.email = email
 self.password = password
 self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 self.socket.connect((self._scrolls_host, self._scrolls_port))
 self.ping_thread = PingThread(self)
 self.message_thread = MessageThread(self)
 self.receive_thread = ReceiveThread(self)
 # self.ping_thread.start()
 self.receive_thread.start()
 self.message_thread.start()
 def login(self):
 login_params = {
 'msg': 'SignIn',
 'email': self._encrypt(self.email),
 'password': self._encrypt(self.password)
 }
 self.send(login_params)
 self.ping_thread.start()
 def subscribe(self, event, callback):
 # add subscribers
 self.subscribers[event] = callback
 def unsubscribe(self, event):
 # rm subscribers
 self.subscribers.pop(event)
 def send(self, params):
 # send message
 self.socket.send(json.dumps(params))
 def receive(self):
 stream_data = ''
 data_json = None
 while (1):
 # read data from the buffer
 data = self.socket.recv(self._socket_recv)
 if not data:
 # no more data being transmitted
 break
 else:
 # append data to the response
 stream_data += data
 try:
 # line breaks means we are handling multiple responses
 if stream_data.find("\n\n"):
 # split and parse each response
 for stream_data_line in stream_data.split("\n\n"):
 # try to load as JSON
 data_json = json.loads(stream_data_line)
 # we have a response, add it to the queue
 self.queue.put(data_json)
 except:
 # invalid json, incomplete data
 pass
 def quit(self):
 # stop all threads and close the socket
 self.receive_thread.stopped = True
 self.receive_thread._Thread__stop()
 self.message_thread.stopped = True
 self.message_thread._Thread__stop()
 self.ping_thread.stopped = True
 self.ping_thread._Thread__stop()
 self.socket.close()
 def _encrypt(self, data):
 key = RSA.importKey(self._scrolls_publickey)
 cipher = PKCS1_v1_5.new(key)
 encrypted_data = cipher.encrypt(data)
 return b64encode(encrypted_data)

Answer 1

Instead of:

Thread.__init__(self)

Do (for example):

super(MessageThread, self).__init__()

Also it's more Pythonic to do:

while True:

Instead of:

while (1):

Instead of using the stopped variables, it may be more efficient to use a variable such as:

self.active = True

That way you can do this which saves you from having to do extra processing on the "not" every iteration:

while self.active:

Even better though may be to just do while True and instead of setting stopped to True and calling Thread_stop(), just call exit() on the threads, then you shouldn't need the stopped variables at all.

Also in receive() you could simplify it slightly

def receive(self):
 stream_data = ''
 while True:
 # read data from the buffer
 data = self.socket.recv(self._socket_recv)
 if not data:
 # no more data being transmitted
 return # now you don't need an else
 # append data to the response
 stream_data += data
 try:
 # line breaks means we are handling multiple responses
 if stream_data.find("\n\n"):
 # split and parse each response
 for stream_data_line in stream_data.split("\n\n"):
 # we have a response, add it to the queue
 # no need to store json.loads result that's only used once
 self.queue.put(json.loads(stream_data_line))
 except:
 # invalid json, incomplete data
 pass

Answer 2

You should read PEP0008, it's the invaluable Python style guide. One such suggestion is to structure your modules so that all the plain import statements are first, separated from the from _ import _ lines. It's neater and easier to read.

import socket
import json
import time
from base64 import b64encode
from threading import Thread
from Queue import Queue
from Crypto.Cipher import PKCS1_v1_5
from Crypto.PublicKey import RSA

Why are you using two different if statements to run the same line? If you need to run it in either case, use or, not an elif.

 if ('msg' in message and message['msg'] == subscriber_key or
 'op' in message and message['op'] == subscriber_key):
 subscriber_callback(message)

If you're going to have a comment under a function declaration, you should make it a docstring so that it can be useful to others reading your code (as docstrings are accessible programmatically through the interpreter).

def subscribe(self, event, callback):
 """Add subscribers"""

Though in that particular case it seems pretty clear what subscribe does to me.

Your receive function has a lot of unnecessary comments, I'd strip them out especially in places that are pretty clear from the actual code. In particular:

 if not data:
 # no more data being transmitted
 break
 else:
 # append data to the response
 stream_data += data

Instead of using str.find(s) use if s in str. It's faster and more Pythonic:

 if "\n\n" in stream_data:

Never use except without giving it a specific exception to look for. If you made a syntax error like a typo then you'd actually not notice because of that except and you'd probably think there was a problem with the JSON. If you're looking for errors from invalid JSON, that raises a ValueError, so raise that instead.

 except ValueError:
 # invalid json, incomplete data
 pass