package corenet import ( "time" context "context" core "github.com/ipfs/go-ipfs/core" net "gx/ipfs/QmQx1dHDDYENugYgqA22BaBrRfuv1coSsuPiM7rYh1wwGH/go-libp2p-net" pro "gx/ipfs/QmZNkThpqfVXs9GNbexPrfBbXSLNYeKrE7jwFM2oqHbyqN/go-libp2p-protocol" pstore "gx/ipfs/QmeXj9VAjmYQZxpmVz7VzccbJrpmr8qkCDSjfVNsPTWTYU/go-libp2p-peerstore" peer "gx/ipfs/QmfMmLGoKzCHDN7cGgk64PJr4iipzidDRME8HABSJqvmhC/go-libp2p-peer" ) type ipfsListener struct { conCh chan net.Stream proto pro.ID ctx context.Context cancel func() } func (il *ipfsListener) Accept() (net.Stream, error) { select { case c := <-il.conCh: return c, nil case <-il.ctx.Done(): return nil, il.ctx.Err() } } func (il *ipfsListener) Close() error { il.cancel() // TODO: unregister handler from peerhost return nil } func Listen(nd *core.IpfsNode, protocol string) (*ipfsListener, error) { ctx, cancel := context.WithCancel(nd.Context()) list := &ipfsListener{ proto: pro.ID(protocol), conCh: make(chan net.Stream), ctx: ctx, cancel: cancel, } nd.PeerHost.SetStreamHandler(list.proto, func(s net.Stream) { select { case list.conCh <- s: case <-ctx.Done(): s.Close() } }) return list, nil } func Dial(nd *core.IpfsNode, p peer.ID, protocol string) (net.Stream, error) { ctx, cancel := context.WithTimeout(nd.Context(), time.Second*30) defer cancel() err := nd.PeerHost.Connect(ctx, pstore.PeerInfo{ID: p}) if err != nil { return nil, err } return nd.PeerHost.NewStream(nd.Context(), p, pro.ID(protocol)) }