varun-r-mallya/py-libp2p
1
0
Fork 0
mirror of https://github.com/varun-r-mallya/py-libp2p.git synced 2026-02-10 23:20:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

[WIP] PubSub and FloodSub development (#133)

Browse Source 
* Add notifee interface

* Add notify function to network interface

* Implement notify feature

* Add tests for notify

* Make notifee functions all async

* Fix linting issue

* Fix linting issue

* Scaffold pubsub router interface

* Scaffold pubsub directory

* Store peer_id in muxed connection

* Implement pubsub notifee

* Remove outdated files

* Implement pubsub first attempt

* Prepare pubsub for floodsub

* Add mplex conn to net stream and add conn in notify tests

* Implement floodsub

* Use NetStream in generic protocol handler

* Debugging async issues

* Modify test to perform proper assert. Test passes

* Remove callbacks. Reduce sleep time

* Add simple three node test

* Clean up code. Add message classes

* Add test for two topics

* Add conn to net stream and conn tests

* Refactor test setup to remove duplicate code

* Fix linting issues

* Fix linting issue

* Fix linting issue

* Fix outstanding unrelated lint issue in multiselect_client

* Add connect function

* Remove debug prints

* Remove debug prints from floodsub

* Use MessageTalk in place of direct message breakdown

* Remove extra prints

* Remove outdated function

* Add message to queues for all topics in message

* Debugging

* Add message self delivery

* Increase read timeout to 5 to get pubsub tests passing

* Refactor testing helper func. Add tests

* Add tests and increase timeout to get tests passing

* Add dummy account demo scaffolding

* Attempt to use threads. Test fails

* Implement basic dummy node tests using threads

* Add generic testing function

* Add simple seven node tree test

* Add more complex seven node tree tests

* Add five node ring tests

* Remove unnecessary get_message_type func

* Add documentation to classes

* Add message id to messages

* Add documentation to test helper func

* Add docs to dummy account node helper func

* Add more docs to dummy account node test helper func

* fixed linting errors in floodsub

* small notify bugfix

* move pubsub into libp2p

* fixed pubsub linting

* fixing pubsub test failures

* linting
This commit is contained in:
stuckinaboot
2019-03-23 13:52:02 -04:00
committed by Stuckinaboot
parent fee905ace2
commit 9c6d441f9f
14 changed files with 1474 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
libp2p/stream_muxer/mplex/mplex.py
View File

@ -11,14 +11,15 @@ class Mplex(IMuxedConn):
reference: https://github.com/libp2p/go-mplex/blob/master/multiplex.go
"""
def __init__(self, conn, generic_protocol_handler):
def __init__(self, conn, generic_protocol_handler, peer_id):
"""
create a new muxed connection
:param conn: an instance of raw connection
:param generic_protocol_handler: generic protocol handler
for new muxed streams
:param peer_id: peer_id of peer the connection is to
"""
super(Mplex, self).__init__(conn, generic_protocol_handler)
super(Mplex, self).__init__(conn, generic_protocol_handler, peer_id)
self.raw_conn = conn
self.initiator = conn.initiator
@ -26,6 +27,9 @@ class Mplex(IMuxedConn):
# Store generic protocol handler
self.generic_protocol_handler = generic_protocol_handler
# Set peer_id
self.peer_id = peer_id
# Mapping from stream ID -> buffer of messages for that stream
self.buffers = {}
@ -56,7 +60,7 @@ class Mplex(IMuxedConn):
# TODO: pass down timeout from user and use that
if stream_id in self.buffers:
try:
data = await asyncio.wait_for(self.buffers[stream_id].get(), timeout=3)
data = await asyncio.wait_for(self.buffers[stream_id].get(), timeout=8)
return data
except asyncio.TimeoutError:
return None