Switch websocket libraries

Author: cpacia

api/restapi.py

@@ -12,6 +12,7 @@
 from twisted.web import server
 from twisted.web.resource import NoResource
 from twisted.web import http
+from twisted.web.server import Site
 from twisted.internet import defer, reactor
 from twisted.protocols.basic import FileSender
 
@@ -1084,3 +1085,16 @@ def get_cases(self, request):
         request.write(json.dumps(cases_list, indent=4))
         request.finish()
         return server.NOT_DONE_YET
+
+
+class RestAPI(Site):
+
+    def __init__(self, mserver, kserver, openbazaar_protocol, only_ip="127.0.0.1", timeout=60 * 60 * 1):
+        self.only_ip = only_ip
+        api_resource = OpenBazaarAPI(mserver, kserver, openbazaar_protocol)
+        Site.__init__(self, api_resource, timeout=timeout)
+
+    def buildProtocol(self, addr):
+        if addr.host != self.only_ip and self.only_ip != "0.0.0.0":
+            return
+        return Site.buildProtocol(self, addr)

api/ws.py

@@ -8,22 +8,26 @@
 from market.profile import Profile
 from keyutils.keys import KeyChain
 from random import shuffle
-from autobahn.twisted.websocket import WebSocketServerFactory, WebSocketServerProtocol
 from protos.countries import CountryCode
 from protos.objects import PlaintextMessage, Value, Listings
 from protos import objects
 from binascii import unhexlify
 from dht.node import Node
+from twisted.internet.protocol import Protocol, Factory, connectionDone
 
 
-class WSProtocol(WebSocketServerProtocol):
+# pylint: disable=W0232
+class WSProtocol(Protocol):
     """
     Handles new incoming requests coming from a websocket.
     """
 
-    def onOpen(self):
+    def connectionMade(self):
         self.factory.register(self)
 
+    def connectionLost(self, reason=connectionDone):
+        self.factory.unregister(self)
+
     def get_vendors(self, message_id):
         if message_id in self.factory.outstanding_vendors:
             queried = self.factory.outstanding_vendors[message_id]
@@ -55,7 +59,7 @@ def handle_response(metadata, node):
                             "nsfw": metadata.nsfw
                         }
                 }
-                self.sendMessage(json.dumps(vendor, indent=4), False)
+                self.transport.write(json.dumps(vendor, indent=4))
                 queried.append(node.id)
                 return True
             else:
@@ -92,7 +96,7 @@ def parse_profile(profile, node):
                                     "fee": profile.moderation_fee
                                 }
                         }
-                        self.sendMessage(json.dumps(moderator, indent=4), False)
+                        self.transport.write(json.dumps(moderator, indent=4))
                     else:
                         m.delete_moderator(node.id)
                 for mod in moderators:
@@ -152,7 +156,7 @@ def handle_response(listings, node):
                                 self.factory.mserver.get_image(node, l.thumbnail_hash)
                             if not os.path.isfile(DATA_FOLDER + 'cache/' + listings.avatar_hash.encode("hex")):
                                 self.factory.mserver.get_image(node, listings.avatar_hash)
-                            self.sendMessage(json.dumps(listing_json, indent=4), False)
+                            self.transport.write(json.dumps(listing_json, indent=4))
                             count += 1
                             self.factory.outstanding_listings[message_id].append(l.contract_hash)
                             if count == 3:
@@ -202,7 +206,7 @@ def respond(l, node):
                 }
                 for country in l.ships_to:
                     listing_json["listing"]["ships_to"].append(str(CountryCode.Name(country)))
-                self.sendMessage(json.dumps(listing_json, indent=4), False)
+                self.transport.write(json.dumps(listing_json, indent=4))
 
         def parse_results(values):
             if values is not None:
@@ -230,7 +234,7 @@ def parse_results(values):
                         pass
         self.factory.kserver.get(keyword.lower()).addCallback(parse_results)
 
-    def onMessage(self, payload, isBinary):
+    def dataReceived(self, payload):
         try:
             request_json = json.loads(payload)
             if isinstance(request_json, unicode):
@@ -262,32 +266,26 @@ def onMessage(self, payload, isBinary):
         except Exception as e:
             print 'Exception occurred: %s' % e
 
-    def connectionLost(self, reason):
-        WebSocketServerProtocol.connectionLost(self, reason)
-        self.factory.unregister(self)
-
-
-class WSFactory(WebSocketServerFactory):
 
-    """
-    Simple broadcast server broadcasting any message it receives to all
-    currently connected clients.
-    """
+class WSFactory(Factory):
 
-    def __init__(self, url, mserver, kserver, only_ip="127.0.0.1", debug=False, debugCodePaths=False):
-        WebSocketServerFactory.__init__(self, url, debug=debug, debugCodePaths=debugCodePaths)
+    def __init__(self, mserver, kserver, only_ip="127.0.0.1"):
         self.mserver = mserver
         self.kserver = kserver
         self.db = mserver.db
         self.outstanding_listings = {}
         self.outstanding_vendors = {}
-        self.clients = []
+        self.protocol = WSProtocol
         self.only_ip = only_ip
+        self.clients = []
+
+    def buildProtocol(self, addr):
+        if addr.host != self.only_ip and self.only_ip != "0.0.0.0":
+            return
+        return Factory.buildProtocol(self, addr)
 
     def register(self, client):
-        if client.transport.getPeer().host != self.only_ip and self.only_ip != "0.0.0.0":
-            client.transport.loseConnection()
-        elif client not in self.clients:
+        if client not in self.clients:
             self.clients.append(client)
 
     def unregister(self, client):
@@ -296,4 +294,6 @@ def unregister(self, client):
 
     def push(self, msg):
         for c in self.clients:
-            c.sendMessage(msg)
+            c.transport.write(msg)
+
+

market/contracts.py

@@ -629,7 +629,7 @@ def add_receipt(self,
                     if s["input_index"] == index:
                         if bitcointools.verify_tx_input(tx, index, redeem_script, s["signature"], vendor_key):
                             tx = bitcointools.apply_multisignatures(tx, index, str(redeem_script),
-                                                               sig, str(s["signature"]))
+                                                                    sig, str(s["signature"]))
                             valid_inputs += 1
             receipt_json["buyer_receipt"]["receipt"]["payout"] = {}
             if valid_inputs == len(outpoints):
@@ -795,7 +795,6 @@ def on_tx_received(self, address_version, address_hash, height, block_hash, tx):
         funding level. We need to keep a running balance and increment it when a new transaction
         is received. If the contract is fully funded, we push a notification to the websockets.
         """
-        print "3"
         try:
             # decode the transaction
             self.log.info("Bitcoin transaction detected")
@@ -1121,8 +1120,8 @@ def validate_for_moderation(self, proof_sig):
             validation_failures.append("Guid signature in vendor_offer not valid;")
 
         valid = bitcointools.ecdsa_raw_verify(listing,
-                                         bitcointools.decode_sig(vendor_bitcoin_signature),
-                                         vendor_bitcoin_pubkey)
+                                              bitcointools.decode_sig(vendor_bitcoin_signature),
+                                              vendor_bitcoin_pubkey)
         if not valid:
             validation_failures.append("Bitcoin signature in vendor_offer is not valid;")
 
@@ -1139,7 +1138,8 @@ def validate_for_moderation(self, proof_sig):
         except Exception:
             validation_failures.append("Guid signature in buyer_order not valid;")
 
-        valid = bitcointools.ecdsa_raw_verify(order, bitcointools.decode_sig(buyer_bitcoin_signature), buyer_bitcoin_pubkey)
+        valid = bitcointools.ecdsa_raw_verify(order, bitcointools.decode_sig(buyer_bitcoin_signature),
+                                              buyer_bitcoin_pubkey)
         if not valid:
             validation_failures.append("Bitcoin signature in buyer_order not valid;")
 

market/network.py

@@ -112,7 +112,8 @@ def get_result(result):
 
                     bitcoin_key = contract["vendor_offer"]["listing"]["id"]["pubkeys"]["bitcoin"]
                     bitcoin_sig = contract["vendor_offer"]["signatures"]["bitcoin"]
-                    valid = bitcointools.ecdsa_raw_verify(verify_obj, bitcointools.decode_sig(bitcoin_sig), bitcoin_key)
+                    valid = bitcointools.ecdsa_raw_verify(verify_obj, bitcointools.decode_sig(bitcoin_sig),
+                                                          bitcoin_key)
                     if not valid:
                         raise Exception("Invalid Bitcoin signature")
 
@@ -901,7 +902,7 @@ def release_funds(self, order_id):
                 if s["input_index"] == index:
                     if bitcointools.verify_tx_input(tx, index, redeem_script, s["signature"], mod_key):
                         tx = bitcointools.apply_multisignatures(tx, index, str(redeem_script),
-                                                           sig, str(s["signature"]))
+                                                                sig, str(s["signature"]))
                         valid_inputs += 1
 
         if valid_inputs == len(outpoints):

openbazaard.py

@@ -8,9 +8,8 @@
 import stun
 import sys
 import time
-from api.ws import WSFactory, WSProtocol
-from api.restapi import OpenBazaarAPI
-from autobahn.twisted.websocket import listenWS
+from api.ws import WSFactory
+from api.restapi import RestAPI
 from constants import DATA_FOLDER, KSIZE, ALPHA, LIBBITCOIN_SERVER,\
     LIBBITCOIN_SERVER_TESTNET, SSL_KEY, SSL_CERT, SEEDS
 from daemon import Daemon
@@ -31,8 +30,7 @@
 from protos.objects import FULL_CONE, RESTRICTED, SYMMETRIC
 from twisted.internet import reactor, task
 from twisted.python import log, logfile
-from twisted.web.server import Site
-from twisted.web.static import File
+from txws import WebSocketFactory
 
 
 def run(*args):
@@ -119,48 +117,22 @@ def on_bootstrap_complete(resp):
 
     reactor.listenUDP(port, protocol)
 
-    class OnlyIP(Site):
-        def __init__(self, resource, ip, timeout=60 * 60 * 1):
-            self.ip = ip
-            Site.__init__(self, resource, timeout=timeout)
-
-        def buildProtocol(self, addr):
-            if addr.host == self.ip:
-                return Site.buildProtocol(self, addr)
-            return None
+    interface = "0.0.0.0" if ALLOWIP not in ("127.0.0.1", "0.0.0.0") else ALLOWIP
 
     # websockets api
+    ws_api = WSFactory(mserver, kserver, only_ip=ALLOWIP)
     if SSL:
-        ws_factory = WSFactory("wss://127.0.0.1:" + str(WSPORT), mserver, kserver, only_ip=ALLOWIP)
-        contextFactory = ChainedOpenSSLContextFactory(SSL_KEY, SSL_CERT)
-        ws_factory.protocol = WSProtocol
-        listenWS(ws_factory, contextFactory)
-    else:
-        ws_factory = WSFactory("ws://127.0.0.1:" + str(WSPORT), mserver, kserver, only_ip=ALLOWIP)
-        ws_factory.protocol = WSProtocol
-        listenWS(ws_factory)
-
-    if ALLOWIP != "127.0.0.1" and ALLOWIP != "0.0.0.0":
-        ws_interface = "0.0.0.0"
+        reactor.listenSSL(RESTPORT, WebSocketFactory(ws_api),
+                          ChainedOpenSSLContextFactory(SSL_KEY, SSL_CERT), interface=interface)
     else:
-        ws_interface = ALLOWIP
-    webdir = File(".")
-    web = Site(webdir)
-
-    reactor.listenTCP(WSPORT - 1, web, interface=ws_interface)
+        reactor.listenTCP(WSPORT, WebSocketFactory(ws_api), interface=interface)
 
     # rest api
-    api = OpenBazaarAPI(mserver, kserver, protocol)
-    if ALLOWIP != "127.0.0.1" and ALLOWIP != "0.0.0.0":
-        rest_interface = "0.0.0.0"
-        site = OnlyIP(api, ALLOWIP, timeout=None)
-    else:
-        rest_interface = ALLOWIP
-        site = Site(api, timeout=None)
+    rest_api = RestAPI(mserver, kserver, protocol, only_ip=ALLOWIP)
     if SSL:
-        reactor.listenSSL(RESTPORT, site, ChainedOpenSSLContextFactory(SSL_KEY, SSL_CERT), interface=rest_interface)
+        reactor.listenSSL(RESTPORT, rest_api, ChainedOpenSSLContextFactory(SSL_KEY, SSL_CERT), interface=interface)
     else:
-        reactor.listenTCP(RESTPORT, site, interface=rest_interface)
+        reactor.listenTCP(RESTPORT, rest_api, interface=interface)
 
     # blockchain
     if TESTNET:
@@ -169,17 +141,19 @@ def buildProtocol(self, addr):
         libbitcoin_client = LibbitcoinClient(LIBBITCOIN_SERVER, log=Logger(service="LibbitcoinClient"))
 
     # listeners
-    nlistener = NotificationListenerImpl(ws_factory, db)
+    nlistener = NotificationListenerImpl(ws_api, db)
     mserver.protocol.add_listener(nlistener)
-    mlistener = MessageListenerImpl(ws_factory, db)
+    mlistener = MessageListenerImpl(ws_api, db)
     mserver.protocol.add_listener(mlistener)
-    blistener = BroadcastListenerImpl(ws_factory, db)
+    blistener = BroadcastListenerImpl(ws_api, db)
     mserver.protocol.add_listener(blistener)
 
-    protocol.set_servers(ws_factory, libbitcoin_client)
+    protocol.set_servers(ws_api, libbitcoin_client)
 
     logger.info("Startup took %s seconds" % str(round(time.time() - args[7], 2)))
 
+    reactor.callLater(15, ws_api.push, "Hello world")
+
     reactor.run()
 
 if __name__ == "__main__":

requirements.txt

@@ -6,8 +6,7 @@ bitcointools==1.1.44
 gnupg==2.0.2
 pynacl==0.3.0
 txrestapi==0.2
-autobahn==0.10.9
-txaio==1.1.0
+txws==0.9.1
 python-libbitcoinclient==0.3.2
 requests==2.7.0
 pyopenssl==0.15.1