import asyncio import uuid import multiaddr from libp2p import new_node from libp2p.host.host_interface import IHost from libp2p.pubsub.floodsub import FloodSub from libp2p.pubsub.pubsub import Pubsub from .configs import FLOODSUB_PROTOCOL_ID from .utils import message_id_generator SUPPORTED_PUBSUB_PROTOCOLS = [FLOODSUB_PROTOCOL_ID] CRYPTO_TOPIC = "ethereum" # Message format: # Sending crypto: ,, # Ex. send,aspyn,alex,5 # Set crypto: , # Ex. set,rob,5 # Determine message type by looking at first item before first comma class DummyAccountNode: """ Node which has an internal balance mapping, meant to serve as a dummy crypto blockchain. There is no actual blockchain, just a simple map indicating how much crypto each user in the mappings holds """ libp2p_node: IHost pubsub: Pubsub floodsub: FloodSub def __init__( self, libp2p_node: IHost, pubsub: Pubsub, floodsub: FloodSub): self.libp2p_node = libp2p_node self.pubsub = pubsub self.floodsub = floodsub self.balances = {} self.next_msg_id_func = message_id_generator(0) self.node_id = str(uuid.uuid1()) @classmethod async def create(cls): """ Create a new DummyAccountNode and attach a libp2p node, a floodsub, and a pubsub instance to this new node We use create as this serves as a factory function and allows us to use async await, unlike the init function """ libp2p_node = await new_node(transport_opt=["/ip4/127.0.0.1/tcp/0"]) await libp2p_node.get_network().listen(multiaddr.Multiaddr("/ip4/127.0.0.1/tcp/0")) floodsub = FloodSub(SUPPORTED_PUBSUB_PROTOCOLS) pubsub = Pubsub( libp2p_node, floodsub, "a", ) return cls( libp2p_node=libp2p_node, pubsub=pubsub, floodsub=floodsub, ) async def handle_incoming_msgs(self): """ Handle all incoming messages on the CRYPTO_TOPIC from peers """ while True: incoming = await self.q.get() msg_comps = incoming.data.decode('utf-8').split(",") if msg_comps[0] == "send": self.handle_send_crypto(msg_comps[1], msg_comps[2], int(msg_comps[3])) elif msg_comps[0] == "set": self.handle_set_crypto(msg_comps[1], int(msg_comps[2])) async def setup_crypto_networking(self): """ Subscribe to CRYPTO_TOPIC and perform call to function that handles all incoming messages on said topic """ self.q = await self.pubsub.subscribe(CRYPTO_TOPIC) asyncio.ensure_future(self.handle_incoming_msgs()) async def publish_send_crypto(self, source_user, dest_user, amount): """ Create a send crypto message and publish that message to all other nodes :param source_user: user to send crypto from :param dest_user: user to send crypto to :param amount: amount of crypto to send """ msg_contents = "send," + source_user + "," + dest_user + "," + str(amount) await self.pubsub.publish(CRYPTO_TOPIC, msg_contents.encode()) async def publish_set_crypto(self, user, amount): """ Create a set crypto message and publish that message to all other nodes :param user: user to set crypto for :param amount: amount of crypto """ msg_contents = "set," + user + "," + str(amount) await self.pubsub.publish(CRYPTO_TOPIC, msg_contents.encode()) def handle_send_crypto(self, source_user, dest_user, amount): """ Handle incoming send_crypto message :param source_user: user to send crypto from :param dest_user: user to send crypto to :param amount: amount of crypto to send """ if source_user in self.balances: self.balances[source_user] -= amount else: self.balances[source_user] = -amount if dest_user in self.balances: self.balances[dest_user] += amount else: self.balances[dest_user] = amount def handle_set_crypto(self, dest_user, amount): """ Handle incoming set_crypto message :param dest_user: user to set crypto for :param amount: amount of crypto """ self.balances[dest_user] = amount def get_balance(self, user): """ Get balance in crypto for a particular user :param user: user to get balance for :return: balance of user """ if user in self.balances: return self.balances[user] else: return -1