From 0ff1abb27983d719c4db6d66f6f174df1d691f6f Mon Sep 17 00:00:00 2001 From: Jeromy Date: Wed, 2 Nov 2016 23:35:03 -0700 Subject: [PATCH] move protocol methods down into peerstream --- package.json | 12 ++++++------ swarm.go | 13 ++++--------- swarm_conn.go | 6 +++++- swarm_stream.go | 33 ++++++++------------------------- 4 files changed, 23 insertions(+), 41 deletions(-) diff --git a/package.json b/package.json index 0cbcb94..dae2324 100644 --- a/package.json +++ b/package.json @@ -9,9 +9,9 @@ "gxDependencies": [ { "author": "whyrusleeping", - "hash": "QmdXimY9QHaasZmw6hWojWnCJvfgxETjZQfg9g6ZrA9wMX", + "hash": "QmZ3qHMAtW9WEHexN3HsjSoduUBbrBPXMEoeddQtqdJUJ6", "name": "go-libp2p-net", - "version": "1.2.1" + "version": "1.3.0" }, { "author": "whyrusleeping", @@ -69,15 +69,15 @@ }, { "author": "whyrusleeping", - "hash": "QmRmFKJgjjQhrT1uDyhpS87kE5M9YbMT8RBWam5uk8o4uH", + "hash": "QmNbLFRGG1uHVfQM2fRFrhc7dgjThXkYUAp5qChFqxYNSH", "name": "go-peerstream", - "version": "1.1.0" + "version": "1.3.0" }, { "author": "whyrusleeping", - "hash": "QmWpTXhTkpoCDEm9twJd5Rc9jFwy61emzxneeJzrVMfjGF", + "hash": "Qmezvofp8DAh8GnU395s2cvim7uACGp6DdiKVftAesEovw", "name": "go-libp2p-metrics", - "version": "1.2.0" + "version": "1.3.0" }, { "author": "whyrusleeping", diff --git a/swarm.go b/swarm.go index bb179bb..ec156d4 100644 --- a/swarm.go +++ b/swarm.go @@ -256,7 +256,7 @@ func (s *Swarm) SetConnHandler(handler ConnHandler) { // See peerstream. func (s *Swarm) SetStreamHandler(handler inet.StreamHandler) { s.swarm.SetStreamHandler(func(s *ps.Stream) { - handler(wrapStream(s)) + handler((*Stream)(s)) }) } @@ -273,12 +273,7 @@ func (s *Swarm) NewStreamWithPeer(ctx context.Context, p peer.ID) (*Stream, erro // TODO: think about passing a context down to NewStreamWithGroup st, err := s.swarm.NewStreamWithGroup(p) - return wrapStream(st), err -} - -// StreamsWithPeer returns all the live Streams to p -func (s *Swarm) StreamsWithPeer(p peer.ID) []*Stream { - return wrapStreams(ps.StreamsWithGroup(p, s.swarm.Streams())) + return (*Stream)(st), err } // ConnectionsToPeer returns all the live connections to p @@ -387,9 +382,9 @@ func (n *ps2netNotifee) Disconnected(c *ps.Conn) { } func (n *ps2netNotifee) OpenedStream(s *ps.Stream) { - n.not.OpenedStream(n.net, &Stream{stream: s}) + n.not.OpenedStream(n.net, (*Stream)(s)) } func (n *ps2netNotifee) ClosedStream(s *ps.Stream) { - n.not.ClosedStream(n.net, &Stream{stream: s}) + n.not.ClosedStream(n.net, (*Stream)(s)) } diff --git a/swarm_conn.go b/swarm_conn.go index 72609f0..bc5b69e 100644 --- a/swarm_conn.go +++ b/swarm_conn.go @@ -77,7 +77,7 @@ func (c *Conn) RemotePublicKey() ic.PubKey { // NewSwarmStream returns a new Stream from this connection func (c *Conn) NewSwarmStream() (*Stream, error) { s, err := c.StreamConn().NewStream() - return wrapStream(s), err + return (*Stream)(s), err } // NewStream returns a new Stream from this connection @@ -91,6 +91,10 @@ func (c *Conn) Close() error { return c.StreamConn().Close() } +func (c *Conn) GetStreams() ([]inet.Stream, error) { + return nil, fmt.Errorf("GetStreams() not yet implemented") +} + func wrapConn(psc *ps.Conn) (*Conn, error) { // grab the underlying connection. if _, ok := psc.NetConn().(iconn.Conn); !ok { diff --git a/swarm_stream.go b/swarm_stream.go index ef3ab3c..e3b1256 100644 --- a/swarm_stream.go +++ b/swarm_stream.go @@ -9,14 +9,11 @@ import ( // Stream is a wrapper around a ps.Stream that exposes a way to get // our Conn and Swarm (instead of just the ps.Conn and ps.Swarm) -type Stream struct { - stream *ps.Stream - protocol protocol.ID -} +type Stream ps.Stream // Stream returns the underlying peerstream.Stream func (s *Stream) Stream() *ps.Stream { - return s.stream + return (*ps.Stream)(s) } // Conn returns the Conn associated with this Stream, as an inet.Conn @@ -26,43 +23,29 @@ func (s *Stream) Conn() inet.Conn { // SwarmConn returns the Conn associated with this Stream, as a *Conn func (s *Stream) SwarmConn() *Conn { - return (*Conn)(s.stream.Conn()) + return (*Conn)(s.Stream().Conn()) } // Read reads bytes from a stream. func (s *Stream) Read(p []byte) (n int, err error) { - return s.stream.Read(p) + return s.Stream().Read(p) } // Write writes bytes to a stream, flushing for each call. func (s *Stream) Write(p []byte) (n int, err error) { - return s.stream.Write(p) + return s.Stream().Write(p) } // Close closes the stream, indicating this side is finished // with the stream. func (s *Stream) Close() error { - return s.stream.Close() + return s.Stream().Close() } func (s *Stream) Protocol() protocol.ID { - return s.protocol + return (*ps.Stream)(s).Protocol() } func (s *Stream) SetProtocol(p protocol.ID) { - s.protocol = p -} - -func wrapStream(pss *ps.Stream) *Stream { - return &Stream{ - stream: pss, - } -} - -func wrapStreams(st []*ps.Stream) []*Stream { - out := make([]*Stream, len(st)) - for i, s := range st { - out[i] = wrapStream(s) - } - return out + (*ps.Stream)(s).SetProtocol(p) } -- GitLab