Skip to content
GitLab
Projects
Groups
Snippets
Help
Loading...
Help
What's new
10
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in
Toggle navigation
Open sidebar
p2p
go-p2p-pubsub
Commits
8676a0e2
Commit
8676a0e2
authored
Jan 12, 2021
by
vyzo
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
deduplicate inbound streams
parent
352c6b99
Changes
2
Hide whitespace changes
Inline
Side-by-side
Showing
2 changed files
with
23 additions
and
1 deletion
+23
-1
comm.go
comm.go
+19
-1
pubsub.go
pubsub.go
+4
-0
No files found.
comm.go
View file @
8676a0e2
...
...
@@ -41,6 +41,17 @@ func (p *PubSub) getHelloPacket() *RPC {
}
func
(
p
*
PubSub
)
handleNewStream
(
s
network
.
Stream
)
{
peer
:=
s
.
Conn
()
.
RemotePeer
()
p
.
inboundStreamsMx
.
Lock
()
other
,
dup
:=
p
.
inboundStreams
[
peer
]
if
dup
{
log
.
Debugf
(
"duplicate inbound stream from %s; resetting other stream"
,
peer
)
other
.
Reset
()
}
p
.
inboundStreams
[
peer
]
=
s
p
.
inboundStreamsMx
.
Unlock
()
r
:=
protoio
.
NewDelimitedReader
(
s
,
p
.
maxMessageSize
)
for
{
rpc
:=
new
(
RPC
)
...
...
@@ -54,10 +65,17 @@ func (p *PubSub) handleNewStream(s network.Stream) {
// but it doesn't hurt to send it.
s
.
Close
()
}
p
.
inboundStreamsMx
.
Lock
()
if
p
.
inboundStreams
[
peer
]
==
s
{
delete
(
p
.
inboundStreams
,
peer
)
}
p
.
inboundStreamsMx
.
Unlock
()
return
}
rpc
.
from
=
s
.
Conn
()
.
RemoteP
eer
()
rpc
.
from
=
p
eer
select
{
case
p
.
incoming
<-
rpc
:
case
<-
p
.
ctx
.
Done
()
:
...
...
pubsub.go
View file @
8676a0e2
...
...
@@ -134,6 +134,9 @@ type PubSub struct {
peers
map
[
peer
.
ID
]
chan
*
RPC
inboundStreamsMx
sync
.
Mutex
inboundStreams
map
[
peer
.
ID
]
network
.
Stream
seenMessagesMx
sync
.
Mutex
seenMessages
*
timecache
.
TimeCache
...
...
@@ -253,6 +256,7 @@ func NewPubSub(ctx context.Context, h host.Host, rt PubSubRouter, opts ...Option
myRelays
:
make
(
map
[
string
]
int
),
topics
:
make
(
map
[
string
]
map
[
peer
.
ID
]
struct
{}),
peers
:
make
(
map
[
peer
.
ID
]
chan
*
RPC
),
inboundStreams
:
make
(
map
[
peer
.
ID
]
network
.
Stream
),
blacklist
:
NewMapBlacklist
(),
blacklistPeer
:
make
(
chan
peer
.
ID
),
seenMessages
:
timecache
.
NewTimeCache
(
TimeCacheDuration
),
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
.
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment