varun-r-mallya/py-libp2p
1
0
Fork 0
mirror of https://github.com/varun-r-mallya/py-libp2p.git synced 2025-12-31 20:36:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

RPC conversion progress

Browse Source
This commit is contained in:
Alex Haynes
2019-03-29 16:23:30 -04:00
committed by Stuckinaboot
parent adce7f0a05
commit 75c12d4aed
2 changed files with 91 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
libp2p/pubsub/floodsub.py
View File

@ -1,4 +1,6 @@
from .pubsub_router_interface import IPubsubRouter
from .pb import rpc_pb2
from .message import MessageSub, MessageTalk
from .message import create_message_talk
class FloodSub(IPubsubRouter):
@ -56,28 +58,49 @@ class FloodSub(IPubsubRouter):
"""
# Encode message
encoded_msg = message.encode()
# encoded_msg = message.encode()
if isinstance(message, str):
msg_talk = create_message_talk(message)
message = rpc_pb2.Message(
from_id=str(msg_talk.origin_id).encode('utf-8'),
seqno=str(msg_talk.message_id).encode('utf-8'),
topicIDs=msg_talk.topics,
data=msg_talk.data.encode()
)
packet = rpc_pb2.RPC()
print("YEET")
print(type(message))
packet.publish.extend([message])
# Get message sender, origin, and topics
msg_talk = create_message_talk(message)
# msg_talk = create_message_talk(message)
msg_sender = str(sender_peer_id)
msg_origin = msg_talk.origin_id
topics = msg_talk.topics
# msg_origin = msg_talk.origin_id
# topics = msg_talk.topics
# Deliver to self if self was origin
# Note: handle_talk checks if self is subscribed to topics in message
if msg_sender == msg_origin and msg_sender == str(self.pubsub.host.get_id()):
await self.pubsub.handle_talk(message)
if msg_sender == message.from_id and msg_sender == str(self.pubsub.host.get_id()):
old_format = MessageTalk(sender_peer_id,
message.from_id,
message.topicIDs,
message.data,
message.seqno)
await self.pubsub.handle_talk(old_format)
# Deliver to self and peers
for topic in topics:
for topic in message.topicIDs:
if topic in self.pubsub.peer_topics:
for peer_id_in_topic in self.pubsub.peer_topics[topic]:
# Forward to all known peers in the topic that are not the
# message sender and are not the message origin
if peer_id_in_topic not in (msg_sender, msg_origin):
if peer_id_in_topic not in (msg_sender, message.from_id):
stream = self.pubsub.peers[peer_id_in_topic]
await stream.write(encoded_msg)
await stream.write(packet.SerializeToString())
else:
# Implies publish did not write
print("publish did not write")