Refactor notifications handling

Author: cpacia

api/restapi.py

@@ -2,23 +2,25 @@
 import json
 import os
 import pickle
+from binascii import unhexlify
+from collections import OrderedDict
+
 from txrestapi.resource import APIResource
 from txrestapi.methods import GET, POST, DELETE
 from twisted.web import server
 from twisted.web.resource import NoResource
 from twisted.web import http
 from twisted.internet import defer, reactor
-from binascii import unhexlify
-from constants import DATA_FOLDER
 from twisted.protocols.basic import FileSender
+
+from constants import DATA_FOLDER
 from protos.countries import CountryCode
 from protos import objects
 from keyutils.keys import KeyChain
 from dht.utils import digest
 from market.profile import Profile
 from market.contracts import Contract
-from collections import OrderedDict
-from upnp import PortMapper
+from net.upnp import PortMapper
 
 DEFAULT_RECORDS_COUNT = 20
 DEFAULT_RECORDS_OFFSET = 0
@@ -538,7 +540,8 @@ def purchase_contract(self, request):
         try:
             def handle_response(resp, contract):
                 if resp:
-                    contract.await_funding(self.protocol.ws, self.protocol.blockchain, resp)
+                    contract.await_funding(self.mserver.protocol.get_notification_listener(),
+                                           self.protocol.blockchain, resp)
                     request.write(json.dumps({"success": True, "payment_address": payment[0],
                                               "amount": payment[1],
                                               "order_id": c.get_contract_id().encode("hex")},
@@ -765,10 +768,12 @@ def get_notifications(self, request):
                 "id": n[0],
                 "guid": n[1].encode("hex"),
                 "handle": n[2],
-                "message": n[3],
-                "timestamp": n[4],
-                "avatar_hash": n[5].encode("hex"),
-                "read": False if n[6] == 0 else True
+                "type": n[3],
+                "order_id": n[4],
+                "title": n[5],
+                "timestamp": n[6],
+                "image_hash": n[7].encode("hex"),
+                "read": False if n[8] == 0 else True
             }
         notification_list.append(notification_json)
         request.write(json.dumps(notification_list, indent=4))
@@ -787,13 +792,13 @@ def mark_as_read(self, request):
             request.finish()
             return server.NOT_DONE_YET
 
-    @POST('^/api/v1/send_notification')
-    def send_notification(self, request):
+    @POST('^/api/v1/broadcast')
+    def broadcast(self, request):
         try:
             def get_response(num):
                 request.write(json.dumps({"success": True, "peers reached": num}, indent=4))
                 request.finish()
-            self.mserver.send_notification(request.args["message"][0]).addCallback(get_response)
+            self.mserver.broadcast(request.args["message"][0]).addCallback(get_response)
             return server.NOT_DONE_YET
         except Exception, e:
             request.write(json.dumps({"success": False, "reason": e.message}, indent=4))

db/datastore.py

@@ -63,8 +63,11 @@ def create_database(filepath=None):
     encryption_pubkey BLOB, subject TEXT, message_type TEXT, message TEXT, timestamp, INTEGER,
     avatar_hash BLOB, signature BLOB, outgoing INTEGER)''')
 
-        cursor.execute('''CREATE TABLE notifications(id TEXT PRIMARY KEY, guid BLOB, handle TEXT, message TEXT,
-    timestamp INTEGER, avatar_hash BLOB, read INTEGER)''')
+        cursor.execute('''CREATE TABLE notifications(id TEXT PRIMARY KEY, guid BLOB, handle TEXT, type TEXT,
+    order_id TEXT, title TEXT, timestamp INTEGER, image_hash BLOB, read INTEGER)''')
+
+        cursor.execute('''CREATE TABLE broadcasts(id TEXT PRIMARY KEY, guid BLOB, handle TEXT, message TEXT,
+    timestamp INTEGER, avatar_hash BLOB)''')
 
         cursor.execute('''CREATE TABLE vendors(guid TEXT PRIMARY KEY, ip TEXT, port INTEGER, signedPubkey BLOB)''')
 
@@ -359,15 +362,17 @@ def __init__(self):
             self.db = lite.connect(DATABASE)
             self.db.text_factory = str
 
-        def save_notification(self, notif_id, guid, handle, message, timestamp, avatar_hash):
+        def save_notification(self, notif_id, guid, handle, notif_type, order_id, title, timestamp, image_hash):
             cursor = self.db.cursor()
-            cursor.execute('''INSERT INTO notifications(id, guid, handle, message, timestamp, avatar_hash, read)
-    VALUES (?,?,?,?,?,?,?)''', (notif_id, guid, handle, message, timestamp, avatar_hash, 0))
+            cursor.execute('''INSERT INTO notifications(id, guid, handle, type, order_id, title, timestamp,
+image_hash, read) VALUES (?,?,?,?,?,?,?,?,?)''', (notif_id, guid, handle, notif_type, order_id, title, timestamp,
+                                                  image_hash, 0))
             self.db.commit()
 
         def get_notifications(self):
             cursor = self.db.cursor()
-            cursor.execute('''SELECT id, guid, handle, message, timestamp, avatar_hash, read FROM notifications''')
+            cursor.execute('''SELECT id, guid, handle, type, order_id, title, timestamp, image_hash, read
+FROM notifications''')
             return cursor.fetchall()
 
         def mark_as_read(self, notif_id):
@@ -380,6 +385,27 @@ def delete_notification(self, notif_id):
             cursor.execute('''DELETE FROM notifications WHERE id=?''', (notif_id,))
             self.db.commit()
 
+    class BroadcastStore(object):
+        def __init__(self):
+            self.db = lite.connect(DATABASE)
+            self.db.text_factory = str
+
+        def save_broadcast(self, broadcast_id, guid, handle, message, timestamp, avatar_hash):
+            cursor = self.db.cursor()
+            cursor.execute('''INSERT INTO broadcasts(id, guid, handle, message, timestamp, avatar_hash)
+    VALUES (?,?,?,?,?,?)''', (broadcast_id, guid, handle, message, timestamp, avatar_hash))
+            self.db.commit()
+
+        def get_broadcasts(self):
+            cursor = self.db.cursor()
+            cursor.execute('''SELECT id, guid, handle, message, timestamp, avatar_hash FROM broadcasts''')
+            return cursor.fetchall()
+
+        def delete_broadcast(self, broadcast_id):
+            cursor = self.db.cursor()
+            cursor.execute('''DELETE FROM broadcasts WHERE id=?''', (broadcast_id,))
+            self.db.commit()
+
     class VendorStore(object):
         def __init__(self):
             self.db = lite.connect(DATABASE)

db/tests/test_datastore.py

@@ -170,8 +170,8 @@ def test_saveMessage(self):
     def test_notificationStore(self):
         n = self.ns.get_notifications()
         self.assertTrue(len(n) == 0)
-        self.ns.save_notification("1234", self.u.guid, self.m.handle, 'NOTICE',
-                                  '0000-00-00 00:00:00', '')
+        self.ns.save_notification("1234", self.u.guid, self.m.handle, 'NOTICE', "", ""
+                                  '0000-00-00 00:00:00', '', 0)
         n = self.ns.get_notifications()
         self.assertIsNotNone(n)
         self.ns.delete_notification("1234")

dht/protocol.py

@@ -4,14 +4,16 @@
 """
 
 import random
+
 from twisted.internet import defer, reactor
 from zope.interface import implements
 import nacl.signing
+
 from dht.node import Node
 from dht.routing import RoutingTable
 from dht.utils import digest
 from log import Logger
-from rpcudp import RPCProtocol
+from net.rpcudp import RPCProtocol
 from interfaces import MessageProcessor
 from protos import objects
 from protos.message import PING, STUN, STORE, DELETE, FIND_NODE, FIND_VALUE, HOLE_PUNCH

dht/tests/test_crawling.py

@@ -7,13 +7,14 @@
 from txrudp import packet, connection, rudp, constants
 from twisted.internet import udp, address, task
 from twisted.trial import unittest
+
 from dht.crawling import RPCFindResponse, NodeSpiderCrawl, ValueSpiderCrawl
 from dht.node import Node, NodeHeap
 from dht.utils import digest
 from dht.storage import ForgetfulStorage
 from dht.protocol import KademliaProtocol
 from protos.objects import Value
-from wireprotocol import OpenBazaarProtocol
+from net.wireprotocol import OpenBazaarProtocol
 from db.datastore import Database
 from constants import ALPHA, KSIZE
 

dht/tests/test_protocol.py

@@ -2,22 +2,23 @@
 import time
 import random
 from binascii import unhexlify
-
 import os
+
 import mock
 import nacl.signing
 import nacl.encoding
 import nacl.hash
 from txrudp import connection, rudp, packet, constants
 from twisted.trial import unittest
 from twisted.internet import task, address, udp, defer, reactor
+
 from dht.protocol import KademliaProtocol
 from dht.utils import digest
 from dht.storage import ForgetfulStorage
 from dht.tests.utils import mknode
 from dht.node import Node
 from protos import message, objects
-from wireprotocol import OpenBazaarProtocol
+from net.wireprotocol import OpenBazaarProtocol
 from db import datastore
 from constants import PROTOCOL_VERSION
 

interfaces.py

@@ -2,6 +2,7 @@
 
 from zope.interface import Interface, Attribute
 
+
 class MessageProcessor(Interface):
     """
     This is an interface for processing messages coming off the wire. Classes that implement this interface should be
@@ -35,17 +36,19 @@ def __iter__():
         :return: iter([list of enums])
         """
 
-class NotificationListener(Interface):
+
+class BroadcastListener(Interface):
     """
-    An interface for handling notifications sent to followers.
+    An interface for handling broadcasts sent to followers.
     """
 
     def notify(guid, message):
         """
-        New notifications will be sent here. They will only show if this node is following the node
-        which sent the notification.
+        New broadcasts will be sent here. They will only show if this node is following the node
+        which sent the broadcast.
         """
 
+
 class MessageListener(Interface):
     """
     An interface for handling messages sent between nodes.
@@ -57,4 +60,24 @@ def notify(plaintext_message, signature):
         Args:
             plaintext_message: the protobuf object containing the message
             signature: the signature covering the message.
+        """
+
+
+class NotificationListener(Interface):
+    """
+    An interface for handling event notifications. New events should update this
+    listener which will save the notifications to the db and push it to UI via websockets.
+    """
+
+    def notify(guid, handle, type, order_id, title, image_hash):
+        """
+        This should be called to register a new notification.
+        Args:
+            guid: (in hex) optional depending on notification type.
+            handle: optional depending on notification type.
+            type: a `String` containing the type of notification,
+                  (ex: Follow, New Order, Order Confirmation, Payment Received).
+            order_id: an order id if this notification is for an order
+            title: a `String` which can be used for the item's title if an order notification.
+            image_hash: optional depending on notification type.
         """