Merge pull request #130 from libp2p/feat/extracting-3

extract net interface and metrics

examples/hosts/main.go

@@ -8,10 +8,10 @@ import (
 	"log"
 	"strings"
 
+	inet "github.com/libp2p/go-libp2p-net"
+	net "github.com/libp2p/go-libp2p-net"
 	host "github.com/libp2p/go-libp2p/p2p/host"
 	bhost "github.com/libp2p/go-libp2p/p2p/host/basic"
-	inet "github.com/libp2p/go-libp2p/p2p/net"
-	net "github.com/libp2p/go-libp2p/p2p/net"
 	swarm "github.com/libp2p/go-libp2p/p2p/net/swarm"
 
 	peer "github.com/ipfs/go-libp2p-peer"

p2p/host/basic/basic_host.go

@@ -5,17 +5,18 @@ import (
 	"io"
 	"time"
 
+	identify "github.com/libp2p/go-libp2p/p2p/protocol/identify"
+	relay "github.com/libp2p/go-libp2p/p2p/protocol/relay"
+
 	peer "github.com/ipfs/go-libp2p-peer"
 	pstore "github.com/ipfs/go-libp2p-peerstore"
 	logging "github.com/ipfs/go-log"
 	ma "github.com/jbenet/go-multiaddr"
 	goprocess "github.com/jbenet/goprocess"
+	metrics "github.com/libp2p/go-libp2p-metrics"
+	mstream "github.com/libp2p/go-libp2p-metrics/stream"
+	inet "github.com/libp2p/go-libp2p-net"
 	protocol "github.com/libp2p/go-libp2p-protocol"
-	metrics "github.com/libp2p/go-libp2p/p2p/metrics"
-	mstream "github.com/libp2p/go-libp2p/p2p/metrics/stream"
-	inet "github.com/libp2p/go-libp2p/p2p/net"
-	identify "github.com/libp2p/go-libp2p/p2p/protocol/identify"
-	relay "github.com/libp2p/go-libp2p/p2p/protocol/relay"
 
 	msmux "github.com/whyrusleeping/go-multistream"
 )

p2p/host/basic/basic_host_test.go

@@ -7,9 +7,9 @@ import (
 	"testing"
 	"time"
 
+	inet "github.com/libp2p/go-libp2p-net"
 	protocol "github.com/libp2p/go-libp2p-protocol"
 	host "github.com/libp2p/go-libp2p/p2p/host"
-	inet "github.com/libp2p/go-libp2p/p2p/net"
 	testutil "github.com/libp2p/go-libp2p/p2p/test/util"
 )
 

p2p/host/basic/natmgr.go

@@ -4,11 +4,12 @@ import (
 	"context"
 	"sync"
 
+	inat "github.com/libp2p/go-libp2p/p2p/nat"
+
 	lgbl "github.com/ipfs/go-libp2p-loggables"
 	ma "github.com/jbenet/go-multiaddr"
 	goprocess "github.com/jbenet/goprocess"
-	inat "github.com/libp2p/go-libp2p/p2p/nat"
-	inet "github.com/libp2p/go-libp2p/p2p/net"
+	inet "github.com/libp2p/go-libp2p-net"
 )
 
 // natManager takes care of adding + removing port mappings to the nat.

p2p/host/host.go

@@ -7,9 +7,9 @@ import (
 	pstore "github.com/ipfs/go-libp2p-peerstore"
 	logging "github.com/ipfs/go-log"
 	ma "github.com/jbenet/go-multiaddr"
+	metrics "github.com/libp2p/go-libp2p-metrics"
+	inet "github.com/libp2p/go-libp2p-net"
 	protocol "github.com/libp2p/go-libp2p-protocol"
-	metrics "github.com/libp2p/go-libp2p/p2p/metrics"
-	inet "github.com/libp2p/go-libp2p/p2p/net"
 	msmux "github.com/whyrusleeping/go-multistream"
 )
 

p2p/host/routed/routed.go

@@ -5,16 +5,16 @@ import (
 	"fmt"
 	"time"
 
+	host "github.com/libp2p/go-libp2p/p2p/host"
+
 	lgbl "github.com/ipfs/go-libp2p-loggables"
 	peer "github.com/ipfs/go-libp2p-peer"
 	pstore "github.com/ipfs/go-libp2p-peerstore"
 	logging "github.com/ipfs/go-log"
 	ma "github.com/jbenet/go-multiaddr"
+	metrics "github.com/libp2p/go-libp2p-metrics"
+	inet "github.com/libp2p/go-libp2p-net"
 	protocol "github.com/libp2p/go-libp2p-protocol"
-	host "github.com/libp2p/go-libp2p/p2p/host"
-	metrics "github.com/libp2p/go-libp2p/p2p/metrics"
-	inet "github.com/libp2p/go-libp2p/p2p/net"
-
 	msmux "github.com/whyrusleeping/go-multistream"
 )
 

p2p/metrics/bw_stats.go

-package metrics
-
-import (
-	gm "github.com/whyrusleeping/go-metrics"
-	"sync"
-
-	peer "github.com/ipfs/go-libp2p-peer"
-	protocol "github.com/libp2p/go-libp2p-protocol"
-)
-
-type Stats struct {
-	TotalIn  int64
-	TotalOut int64
-	RateIn   float64
-	RateOut  float64
-}
-
-type BandwidthCounter struct {
-	lock     sync.Mutex
-	totalIn  gm.Meter
-	totalOut gm.Meter
-	reg      gm.Registry
-}
-
-func NewBandwidthCounter() *BandwidthCounter {
-	reg := gm.NewRegistry()
-	return &BandwidthCounter{
-		totalIn:  gm.GetOrRegisterMeter("totalIn", reg),
-		totalOut: gm.GetOrRegisterMeter("totalOut", reg),
-		reg:      reg,
-	}
-}
-
-func (bwc *BandwidthCounter) LogSentMessage(size int64) {
-	bwc.totalOut.Mark(size)
-}
-
-func (bwc *BandwidthCounter) LogRecvMessage(size int64) {
-	bwc.totalIn.Mark(size)
-}
-
-func (bwc *BandwidthCounter) LogSentMessageStream(size int64, proto protocol.ID, p peer.ID) {
-	meter := gm.GetOrRegisterMeter("/peer/out/"+string(p), bwc.reg)
-	meter.Mark(size)
-
-	pmeter := gm.GetOrRegisterMeter("/proto/out/"+string(proto), bwc.reg)
-	pmeter.Mark(size)
-}
-
-func (bwc *BandwidthCounter) LogRecvMessageStream(size int64, proto protocol.ID, p peer.ID) {
-	meter := gm.GetOrRegisterMeter("/peer/in/"+string(p), bwc.reg)
-	meter.Mark(size)
-
-	pmeter := gm.GetOrRegisterMeter("/proto/in/"+string(proto), bwc.reg)
-	pmeter.Mark(size)
-}
-
-func (bwc *BandwidthCounter) GetBandwidthForPeer(p peer.ID) (out Stats) {
-	inMeter := gm.GetOrRegisterMeter("/peer/in/"+string(p), bwc.reg).Snapshot()
-	outMeter := gm.GetOrRegisterMeter("/peer/out/"+string(p), bwc.reg).Snapshot()
-
-	return Stats{
-		TotalIn:  inMeter.Count(),
-		TotalOut: outMeter.Count(),
-		RateIn:   inMeter.RateFine(),
-		RateOut:  outMeter.RateFine(),
-	}
-}
-
-func (bwc *BandwidthCounter) GetBandwidthForProtocol(proto protocol.ID) (out Stats) {
-	inMeter := gm.GetOrRegisterMeter(string("/proto/in/"+proto), bwc.reg).Snapshot()
-	outMeter := gm.GetOrRegisterMeter(string("/proto/out/"+proto), bwc.reg).Snapshot()
-
-	return Stats{
-		TotalIn:  inMeter.Count(),
-		TotalOut: outMeter.Count(),
-		RateIn:   inMeter.RateFine(),
-		RateOut:  outMeter.RateFine(),
-	}
-}
-
-func (bwc *BandwidthCounter) GetBandwidthTotals() (out Stats) {
-	return Stats{
-		TotalIn:  bwc.totalIn.Count(),
-		TotalOut: bwc.totalOut.Count(),
-		RateIn:   bwc.totalIn.RateFine(),
-		RateOut:  bwc.totalOut.RateFine(),
-	}
-}

p2p/metrics/conn/conn.go

-package meterconn
-
-import (
-	transport "github.com/libp2p/go-libp2p-transport"
-	metrics "github.com/libp2p/go-libp2p/p2p/metrics"
-)
-
-type MeteredConn struct {
-	mesRecv metrics.MeterCallback
-	mesSent metrics.MeterCallback
-
-	transport.Conn
-}
-
-func WrapConn(bwc metrics.Reporter, c transport.Conn) transport.Conn {
-	return newMeteredConn(c, bwc.LogRecvMessage, bwc.LogSentMessage)
-}
-
-func newMeteredConn(base transport.Conn, rcb metrics.MeterCallback, scb metrics.MeterCallback) transport.Conn {
-	return &MeteredConn{
-		Conn:    base,
-		mesRecv: rcb,
-		mesSent: scb,
-	}
-}
-
-func (mc *MeteredConn) Read(b []byte) (int, error) {
-	n, err := mc.Conn.Read(b)
-
-	mc.mesRecv(int64(n))
-	return n, err
-}
-
-func (mc *MeteredConn) Write(b []byte) (int, error) {
-	n, err := mc.Conn.Write(b)
-
-	mc.mesSent(int64(n))
-	return n, err
-}

p2p/metrics/interface.go

-package metrics
-
-import (
-	peer "github.com/ipfs/go-libp2p-peer"
-	protocol "github.com/libp2p/go-libp2p-protocol"
-)
-
-type StreamMeterCallback func(int64, protocol.ID, peer.ID)
-type MeterCallback func(int64)
-
-type Reporter interface {
-	LogSentMessage(int64)
-	LogRecvMessage(int64)
-	LogSentMessageStream(int64, protocol.ID, peer.ID)
-	LogRecvMessageStream(int64, protocol.ID, peer.ID)
-	GetBandwidthForPeer(peer.ID) Stats
-	GetBandwidthForProtocol(protocol.ID) Stats
-	GetBandwidthTotals() Stats
-}

p2p/metrics/stream/metered.go

-package meterstream
-
-import (
-	peer "github.com/ipfs/go-libp2p-peer"
-	protocol "github.com/libp2p/go-libp2p-protocol"
-	metrics "github.com/libp2p/go-libp2p/p2p/metrics"
-	inet "github.com/libp2p/go-libp2p/p2p/net"
-)
-
-type meteredStream struct {
-	// keys for accessing metrics data
-	protoKey protocol.ID
-	peerKey  peer.ID
-
-	inet.Stream
-
-	// callbacks for reporting bandwidth usage
-	mesSent metrics.StreamMeterCallback
-	mesRecv metrics.StreamMeterCallback
-}
-
-func newMeteredStream(base inet.Stream, p peer.ID, recvCB, sentCB metrics.StreamMeterCallback) inet.Stream {
-	return &meteredStream{
-		Stream:   base,
-		mesSent:  sentCB,
-		mesRecv:  recvCB,
-		protoKey: base.Protocol(),
-		peerKey:  p,
-	}
-}
-
-func WrapStream(base inet.Stream, bwc metrics.Reporter) inet.Stream {
-	return newMeteredStream(base, base.Conn().RemotePeer(), bwc.LogRecvMessageStream, bwc.LogSentMessageStream)
-}
-
-func (s *meteredStream) Read(b []byte) (int, error) {
-	n, err := s.Stream.Read(b)
-
-	// Log bytes read
-	s.mesRecv(int64(n), s.protoKey, s.peerKey)
-
-	return n, err
-}
-
-func (s *meteredStream) Write(b []byte) (int, error) {
-	n, err := s.Stream.Write(b)
-
-	// Log bytes written
-	s.mesSent(int64(n), s.protoKey, s.peerKey)
-
-	return n, err
-}

p2p/metrics/stream/metered_test.go

-package meterstream
-
-import (
-	"io"
-	"io/ioutil"
-	"testing"
-
-	randbo "github.com/dustin/randbo"
-	peer "github.com/ipfs/go-libp2p-peer"
-	protocol "github.com/libp2p/go-libp2p-protocol"
-	inet "github.com/libp2p/go-libp2p/p2p/net"
-)
-
-type FakeStream struct {
-	ReadBuf io.Reader
-	inet.Stream
-}
-
-func (fs *FakeStream) Read(b []byte) (int, error) {
-	return fs.ReadBuf.Read(b)
-}
-
-func (fs *FakeStream) Write(b []byte) (int, error) {
-	return len(b), nil
-}
-
-func (fs *FakeStream) Protocol() protocol.ID {
-	return "TEST"
-}
-
-func TestCallbacksWork(t *testing.T) {
-	fake := new(FakeStream)
-
-	var sent int64
-	var recv int64
-
-	sentCB := func(n int64, proto protocol.ID, p peer.ID) {
-		sent += n
-	}
-
-	recvCB := func(n int64, proto protocol.ID, p peer.ID) {
-		recv += n
-	}
-
-	ms := newMeteredStream(fake, peer.ID("PEER"), recvCB, sentCB)
-
-	toWrite := int64(100000)
-	toRead := int64(100000)
-
-	fake.ReadBuf = io.LimitReader(randbo.New(), toRead)
-	writeData := io.LimitReader(randbo.New(), toWrite)
-
-	n, err := io.Copy(ms, writeData)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	if n != toWrite {
-		t.Fatal("incorrect write amount")
-	}
-
-	if toWrite != sent {
-		t.Fatal("incorrectly reported writes", toWrite, sent)
-	}
-
-	n, err = io.Copy(ioutil.Discard, ms)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	if n != toRead {
-		t.Fatal("incorrect read amount")
-	}
-
-	if toRead != recv {
-		t.Fatal("incorrectly reported reads")
-	}
-}

p2p/net/interface.go

-package net
-
-import (
-	"context"
-	"io"
-
-	peer "github.com/ipfs/go-libp2p-peer"
-	pstore "github.com/ipfs/go-libp2p-peerstore"
-	ma "github.com/jbenet/go-multiaddr"
-	"github.com/jbenet/goprocess"
-	conn "github.com/libp2p/go-libp2p-conn"
-	protocol "github.com/libp2p/go-libp2p-protocol"
-)
-
-// MessageSizeMax is a soft (recommended) maximum for network messages.
-// One can write more, as the interface is a stream. But it is useful
-// to bunch it up into multiple read/writes when the whole message is
-// a single, large serialized object.
-const MessageSizeMax = 2 << 22 // 4MB
-
-// Stream represents a bidirectional channel between two agents in
-// the IPFS network. "agent" is as granular as desired, potentially
-// being a "request -> reply" pair, or whole protocols.
-// Streams are backed by SPDY streams underneath the hood.
-type Stream interface {
-	io.Reader
-	io.Writer
-	io.Closer
-
-	Protocol() protocol.ID
-	SetProtocol(protocol.ID)
-
-	// Conn returns the connection this stream is part of.
-	Conn() Conn
-}
-
-// StreamHandler is the type of function used to listen for
-// streams opened by the remote side.
-type StreamHandler func(Stream)
-
-// Conn is a connection to a remote peer. It multiplexes streams.
-// Usually there is no need to use a Conn directly, but it may
-// be useful to get information about the peer on the other side:
-//  stream.Conn().RemotePeer()
-type Conn interface {
-	conn.PeerConn
-
-	// NewStream constructs a new Stream over this conn.
-	NewStream() (Stream, error)
-}
-
-// ConnHandler is the type of function used to listen for
-// connections opened by the remote side.
-type ConnHandler func(Conn)
-
-// Network is the interface used to connect to the outside world.
-// It dials and listens for connections. it uses a Swarm to pool
-// connnections (see swarm pkg, and peerstream.Swarm). Connections
-// are encrypted with a TLS-like protocol.
-type Network interface {
-	Dialer
-	io.Closer
-
-	// SetStreamHandler sets the handler for new streams opened by the
-	// remote side. This operation is threadsafe.
-	SetStreamHandler(StreamHandler)
-
-	// SetConnHandler sets the handler for new connections opened by the
-	// remote side. This operation is threadsafe.
-	SetConnHandler(ConnHandler)
-
-	// NewStream returns a new stream to given peer p.
-	// If there is no connection to p, attempts to create one.
-	NewStream(context.Context, peer.ID) (Stream, error)
-
-	// Listen tells the network to start listening on given multiaddrs.
-	Listen(...ma.Multiaddr) error
-
-	// ListenAddresses returns a list of addresses at which this network listens.
-	ListenAddresses() []ma.Multiaddr
-
-	// InterfaceListenAddresses returns a list of addresses at which this network
-	// listens. It expands "any interface" addresses (/ip4/0.0.0.0, /ip6/::) to
-	// use the known local interfaces.
-	InterfaceListenAddresses() ([]ma.Multiaddr, error)
-
-	// Process returns the network's Process
-	Process() goprocess.Process
-}
-
-// Dialer represents a service that can dial out to peers
-// (this is usually just a Network, but other services may not need the whole
-// stack, and thus it becomes easier to mock)
-type Dialer interface {
-
-	// Peerstore returns the internal peerstore
-	// This is useful to tell the dialer about a new address for a peer.
-	// Or use one of the public keys found out over the network.
-	Peerstore() pstore.Peerstore
-
-	// LocalPeer returns the local peer associated with this network
-	LocalPeer() peer.ID
-
-	// DialPeer establishes a connection to a given peer
-	DialPeer(context.Context, peer.ID) (Conn, error)
-
-	// ClosePeer closes the connection to a given peer
-	ClosePeer(peer.ID) error
-
-	// Connectedness returns a state signaling connection capabilities
-	Connectedness(peer.ID) Connectedness
-
-	// Peers returns the peers connected
-	Peers() []peer.ID
-
-	// Conns returns the connections in this Netowrk
-	Conns() []Conn
-
-	// ConnsToPeer returns the connections in this Netowrk for given peer.
-	ConnsToPeer(p peer.ID) []Conn
-
-	// Notify/StopNotify register and unregister a notifiee for signals
-	Notify(Notifiee)
-	StopNotify(Notifiee)
-}
-
-// Connectedness signals the capacity for a connection with a given node.
-// It is used to signal to services and other peers whether a node is reachable.
-type Connectedness int
-
-const (
-	// NotConnected means no connection to peer, and no extra information (default)
-	NotConnected Connectedness = iota
-
-	// Connected means has an open, live connection to peer
-	Connected
-
-	// CanConnect means recently connected to peer, terminated gracefully
-	CanConnect
-
-	// CannotConnect means recently attempted connecting but failed to connect.
-	// (should signal "made effort, failed")
-	CannotConnect
-)
-
-// Notifiee is an interface for an object wishing to receive
-// notifications from a Network.
-type Notifiee interface {
-	Listen(Network, ma.Multiaddr)      // called when network starts listening on an addr
-	ListenClose(Network, ma.Multiaddr) // called when network starts listening on an addr
-	Connected(Network, Conn)           // called when a connection opened
-	Disconnected(Network, Conn)        // called when a connection closed
-	OpenedStream(Network, Stream)      // called when a stream opened
-	ClosedStream(Network, Stream)      // called when a stream closed
-
-	// TODO
-	// PeerConnected(Network, peer.ID)    // called when a peer connected
-	// PeerDisconnected(Network, peer.ID) // called when a peer disconnected
-}

p2p/net/mock/interface.go

@@ -11,12 +11,12 @@ import (
 	"time"
 
 	host "github.com/libp2p/go-libp2p/p2p/host"
-	inet "github.com/libp2p/go-libp2p/p2p/net"
 
 	ic "github.com/ipfs/go-libp2p-crypto"
 	peer "github.com/ipfs/go-libp2p-peer"
 	pstore "github.com/ipfs/go-libp2p-peerstore"
 	ma "github.com/jbenet/go-multiaddr"
+	inet "github.com/libp2p/go-libp2p-net"
 )
 
 type Mocknet interface {

p2p/net/mock/mock_conn.go

@@ -8,7 +8,7 @@ import (
 	peer "github.com/ipfs/go-libp2p-peer"
 	ma "github.com/jbenet/go-multiaddr"
 	process "github.com/jbenet/goprocess"
-	inet "github.com/libp2p/go-libp2p/p2p/net"
+	inet "github.com/libp2p/go-libp2p-net"
 )
 
 // conn represents one side's perspective of a

p2p/net/mock/mock_link.go

@@ -7,7 +7,7 @@ import (
 	"time"
 
 	peer "github.com/ipfs/go-libp2p-peer"
-	inet "github.com/libp2p/go-libp2p/p2p/net"
+	inet "github.com/libp2p/go-libp2p-net"
 )
 
 // link implements mocknet.Link

p2p/net/mock/mock_net.go

@@ -8,7 +8,6 @@ import (
 
 	host "github.com/libp2p/go-libp2p/p2p/host"
 	bhost "github.com/libp2p/go-libp2p/p2p/host/basic"
-	inet "github.com/libp2p/go-libp2p/p2p/net"
 	p2putil "github.com/libp2p/go-libp2p/p2p/test/util"
 
 	ic "github.com/ipfs/go-libp2p-crypto"
@@ -17,6 +16,7 @@ import (
 	ma "github.com/jbenet/go-multiaddr"
 	"github.com/jbenet/goprocess"
 	goprocessctx "github.com/jbenet/goprocess/context"
+	inet "github.com/libp2p/go-libp2p-net"
 	testutil "github.com/libp2p/go-testutil"
 )
 

p2p/net/mock/mock_notif_test.go

@@ -7,7 +7,7 @@ import (
 
 	peer "github.com/ipfs/go-libp2p-peer"
 	ma "github.com/jbenet/go-multiaddr"
-	inet "github.com/libp2p/go-libp2p/p2p/net"
+	inet "github.com/libp2p/go-libp2p-net"
 )
 
 func TestNotifications(t *testing.T) {

p2p/net/mock/mock_peernet.go

 package mocknet
 
 import (
+	"context"
 	"fmt"
 	"math/rand"
 	"sync"
 
-	inet "github.com/libp2p/go-libp2p/p2p/net"
-
-	"context"
 	peer "github.com/ipfs/go-libp2p-peer"
 	pstore "github.com/ipfs/go-libp2p-peerstore"
 	ma "github.com/jbenet/go-multiaddr"
 	"github.com/jbenet/goprocess"
 	goprocessctx "github.com/jbenet/goprocess/context"
+	inet "github.com/libp2p/go-libp2p-net"
 )
 
 // peernet implements inet.Network

p2p/net/mock/mock_printer.go

@@ -5,7 +5,7 @@ import (
 	"io"
 
 	peer "github.com/ipfs/go-libp2p-peer"
-	inet "github.com/libp2p/go-libp2p/p2p/net"
+	inet "github.com/libp2p/go-libp2p-net"
 )
 
 // separate object so our interfaces are separate :)

p2p/net/mock/mock_stream.go

@@ -6,8 +6,8 @@ import (
 	"time"
 
 	process "github.com/jbenet/goprocess"
+	inet "github.com/libp2p/go-libp2p-net"
 	protocol "github.com/libp2p/go-libp2p-protocol"
-	inet "github.com/libp2p/go-libp2p/p2p/net"
 )
 
 // stream implements inet.Stream