Browse Source
p2p: implement new Transport interface (#5791)
This implements a new `Transport` interface and related types for the P2P refactor in #5670. Previously, `conn.MConnection` was very tightly coupled to the `Peer` implementation -- in order to allow alternative non-multiplexed transports (e.g. QUIC), MConnection has now been moved below the `Transport` interface, as `MConnTransport`, and decoupled from the peer. Since the `p2p` package is not covered by our Go API stability, this is not considered a breaking change, and not listed in the changelog. The initial approach was to implement the new interface in its final form (which also involved possible protocol changes, see https://github.com/tendermint/spec/pull/227). However, it turned out that this would require a large amount of changes to existing P2P code because of the previous tight coupling between `Peer` and `MConnection` and the reliance on subtleties in the MConnection behavior. Instead, I have broadened the `Transport` interface to expose much of the existing MConnection interface, preserved much of the existing MConnection logic and behavior in the transport implementation, and tried to make as few changes to the rest of the P2P stack as possible. We will instead reduce this interface gradually as we refactor other parts of the P2P stack. The low-level transport code and protocol (e.g. MConnection, SecretConnection and so on) has not been significantly changed, and refactoring this is not a priority until we come up with a plan for QUIC adoption, as we may end up discarding the MConnection code entirely. There are no tests of the new `MConnTransport`, as this code is likely to evolve as we proceed with the P2P refactor, but tests should be added before a final release. The E2E tests are sufficient for basic validation in the meanwhile.pull/5799/head
Erik Grinaker
4 years ago
committed by
GitHub
GitHub
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
16 changed files with 1098 additions and 1591 deletions
Split View
Diff Options
-
+24 -41consensus/byzantine_test.go
-
+7 -1libs/log/testing_logger.go
-
+26 -27node/node.go
-
+2 -7p2p/conn/connection.go
-
+10 -0p2p/netaddress.go
-
+1 -1p2p/node_info.go
-
+75 -108p2p/peer.go
-
+13 -23p2p/peer_test.go
-
+13 -8p2p/pex/pex_reactor_test.go
-
+45 -23p2p/switch.go
-
+40 -35p2p/switch_test.go
-
+12 -33p2p/test_util.go
-
+129 -561p2p/transport.go
-
+675 -0p2p/transport_mconn.go
-
+0 -696p2p/transport_test.go
-
+26 -27test/maverick/node/node.go
+ 24
- 41
consensus/byzantine_test.go
View File
+ 7
- 1
libs/log/testing_logger.go
View File
+ 26
- 27
node/node.go
View File
+ 2
- 7
p2p/conn/connection.go
View File
+ 10
- 0
p2p/netaddress.go
View File
+ 1
- 1
p2p/node_info.go
View File
+ 75
- 108
p2p/peer.go
View File
+ 13
- 23
p2p/peer_test.go
View File
+ 13
- 8
p2p/pex/pex_reactor_test.go
View File
+ 45
- 23
p2p/switch.go
View File
+ 40
- 35
p2p/switch_test.go
View File
+ 12
- 33
p2p/test_util.go
View File
+ 129
- 561
p2p/transport.go
View File
+ 675
- 0
p2p/transport_mconn.go
View File
| @ -0,0 +1,675 @@ | |||
| package p2p | |||
| import ( | |||
| "context" | |||
| "errors" | |||
| "fmt" | |||
| "io" | |||
| "net" | |||
| "sync" | |||
| "time" | |||
| "github.com/tendermint/tendermint/crypto" | |||
| "github.com/tendermint/tendermint/libs/log" | |||
| "github.com/tendermint/tendermint/libs/protoio" | |||
| "github.com/tendermint/tendermint/p2p/conn" | |||
| p2pproto "github.com/tendermint/tendermint/proto/tendermint/p2p" | |||
| "golang.org/x/net/netutil" | |||
| ) | |||
| const ( | |||
| defaultDialTimeout = time.Second | |||
| defaultFilterTimeout = 5 * time.Second | |||
| defaultHandshakeTimeout = 3 * time.Second | |||
| ) | |||
| // MConnProtocol is the MConn protocol identifier. | |||
| const MConnProtocol Protocol = "mconn" | |||
| // MConnTransportOption sets an option for MConnTransport. | |||
| type MConnTransportOption func(*MConnTransport) | |||
| // MConnTransportMaxIncomingConnections sets the maximum number of | |||
| // simultaneous incoming connections. Default: 0 (unlimited) | |||
| func MConnTransportMaxIncomingConnections(max int) MConnTransportOption { | |||
| return func(mt *MConnTransport) { mt.maxIncomingConnections = max } | |||
| } | |||
| // MConnTransportFilterTimeout sets the timeout for filter callbacks. | |||
| func MConnTransportFilterTimeout(timeout time.Duration) MConnTransportOption { | |||
| return func(mt *MConnTransport) { mt.filterTimeout = timeout } | |||
| } | |||
| // MConnTransportConnFilters sets connection filters. | |||
| func MConnTransportConnFilters(filters ...ConnFilterFunc) MConnTransportOption { | |||
| return func(mt *MConnTransport) { mt.connFilters = filters } | |||
| } | |||
| // ConnFilterFunc is a callback for connection filtering. If it returns an | |||
| // error, the connection is rejected. The set of existing connections is passed | |||
| // along with the new connection and all resolved IPs. | |||
| type ConnFilterFunc func(ConnSet, net.Conn, []net.IP) error | |||
| // ConnDuplicateIPFilter resolves and keeps all ips for an incoming connection | |||
| // and refuses new ones if they come from a known ip. | |||
| var ConnDuplicateIPFilter ConnFilterFunc = func(cs ConnSet, c net.Conn, ips []net.IP) error { | |||
| for _, ip := range ips { | |||
| if cs.HasIP(ip) { | |||
| return ErrRejected{ | |||
| conn: c, | |||
| err: fmt.Errorf("ip<%v> already connected", ip), | |||
| isDuplicate: true, | |||
| } | |||
| } | |||
| } | |||
| return nil | |||
| } | |||
| // MConnTransport is a Transport implementation using the current multiplexed | |||
| // Tendermint protocol ("MConn"). It inherits lots of code and logic from the | |||
| // previous implementation for parity with the current P2P stack (such as | |||
| // connection filtering, peer verification, and panic handling), which should be | |||
| // moved out of the transport once the rest of the P2P stack is rewritten. | |||
| type MConnTransport struct { | |||
| privKey crypto.PrivKey | |||
| nodeInfo DefaultNodeInfo | |||
| channelDescs []*ChannelDescriptor | |||
| mConnConfig conn.MConnConfig | |||
| maxIncomingConnections int | |||
| dialTimeout time.Duration | |||
| handshakeTimeout time.Duration | |||
| filterTimeout time.Duration | |||
| logger log.Logger | |||
| listener net.Listener | |||
| closeOnce sync.Once | |||
| chAccept chan *mConnConnection | |||
| chError chan error | |||
| chClose chan struct{} | |||
| // FIXME: This is a vestige from the old transport, and should be managed | |||
| // by the router once we rewrite the P2P core. | |||
| conns ConnSet | |||
| connFilters []ConnFilterFunc | |||
| } | |||
| // NewMConnTransport sets up a new MConn transport. | |||
| func NewMConnTransport( | |||
| logger log.Logger, | |||
| nodeInfo NodeInfo, // FIXME: should use DefaultNodeInfo, left for code compatibility | |||
| privKey crypto.PrivKey, | |||
| mConnConfig conn.MConnConfig, | |||
| opts ...MConnTransportOption, | |||
| ) *MConnTransport { | |||
| m := &MConnTransport{ | |||
| privKey: privKey, | |||
| nodeInfo: nodeInfo.(DefaultNodeInfo), | |||
| mConnConfig: mConnConfig, | |||
| channelDescs: []*ChannelDescriptor{}, | |||
| dialTimeout: defaultDialTimeout, | |||
| handshakeTimeout: defaultHandshakeTimeout, | |||
| filterTimeout: defaultFilterTimeout, | |||
| logger: logger, | |||
| chAccept: make(chan *mConnConnection), | |||
| chError: make(chan error), | |||
| chClose: make(chan struct{}), | |||
| conns: NewConnSet(), | |||
| connFilters: []ConnFilterFunc{}, | |||
| } | |||
| for _, opt := range opts { | |||
| opt(m) | |||
| } | |||
| return m | |||
| } | |||
| // SetChannelDescriptors implements Transport. | |||
| func (m *MConnTransport) SetChannelDescriptors(chDescs []*conn.ChannelDescriptor) { | |||
| m.channelDescs = chDescs | |||
| } | |||
| // Listen asynchronously listens for inbound connections on the given endpoint. | |||
| // It must be called exactly once before calling Accept(), and the caller must | |||
| // call Close() to shut down the listener. | |||
| func (m *MConnTransport) Listen(endpoint Endpoint) error { | |||
| if m.listener != nil { | |||
| return errors.New("MConn transport is already listening") | |||
| } | |||
| err := m.normalizeEndpoint(&endpoint) | |||
| if err != nil { | |||
| return fmt.Errorf("invalid MConn listen endpoint %q: %w", endpoint, err) | |||
| } | |||
| m.listener, err = net.Listen("tcp", fmt.Sprintf("%v:%v", endpoint.IP, endpoint.Port)) | |||
| if err != nil { | |||
| return err | |||
| } | |||
| if m.maxIncomingConnections > 0 { | |||
| m.listener = netutil.LimitListener(m.listener, m.maxIncomingConnections) | |||
| } | |||
| // Spawn a goroutine to accept inbound connections asynchronously. | |||
| go m.accept() | |||
| return nil | |||
| } | |||
| // accept accepts inbound connections in a loop, and asynchronously handshakes | |||
| // with the peer to avoid head-of-line blocking. Established connections are | |||
| // passed to Accept() via chAccept. | |||
| // See: https://github.com/tendermint/tendermint/issues/204 | |||
| func (m *MConnTransport) accept() { | |||
| for { | |||
| tcpConn, err := m.listener.Accept() | |||
| if err != nil { | |||
| // We have to check for closure first, since we don't want to | |||
| // propagate "use of closed network connection" errors. | |||
| select { | |||
| case <-m.chClose: | |||
| default: | |||
| // We also select on chClose here, in case the transport closes | |||
| // while we're blocked on error propagation. | |||
| select { | |||
| case m.chError <- err: | |||
| case <-m.chClose: | |||
| } | |||
| } | |||
| return | |||
| } | |||
| go func() { | |||
| err := m.filterTCPConn(tcpConn) | |||
| if err != nil { | |||
| if err := tcpConn.Close(); err != nil { | |||
| m.logger.Debug("failed to close TCP connection", "err", err) | |||
| } | |||
| select { | |||
| case m.chError <- err: | |||
| case <-m.chClose: | |||
| } | |||
| return | |||
| } | |||
| conn, err := newMConnConnection(m, tcpConn, "") | |||
| if err != nil { | |||
| m.conns.Remove(tcpConn) | |||
| if err := tcpConn.Close(); err != nil { | |||
| m.logger.Debug("failed to close TCP connection", "err", err) | |||
| } | |||
| select { | |||
| case m.chError <- err: | |||
| case <-m.chClose: | |||
| } | |||
| } else { | |||
| select { | |||
| case m.chAccept <- conn: | |||
| case <-m.chClose: | |||
| if err := tcpConn.Close(); err != nil { | |||
| m.logger.Debug("failed to close TCP connection", "err", err) | |||
| } | |||
| } | |||
| } | |||
| }() | |||
| } | |||
| } | |||
| // Accept implements Transport. | |||
| // | |||
| // accept() runs a concurrent accept loop that accepts inbound connections | |||
| // and then handshakes in a non-blocking fashion. The handshaked and validated | |||
| // connections are returned via this call, picking them off of the chAccept | |||
| // channel (or the handshake error, if any). | |||
| func (m *MConnTransport) Accept(ctx context.Context) (Connection, error) { | |||
| select { | |||
| case conn := <-m.chAccept: | |||
| return conn, nil | |||
| case err := <-m.chError: | |||
| return nil, err | |||
| case <-m.chClose: | |||
| return nil, ErrTransportClosed{} | |||
| case <-ctx.Done(): | |||
| return nil, nil | |||
| } | |||
| } | |||
| // Dial implements Transport. | |||
| func (m *MConnTransport) Dial(ctx context.Context, endpoint Endpoint) (Connection, error) { | |||
| err := m.normalizeEndpoint(&endpoint) | |||
| if err != nil { | |||
| return nil, err | |||
| } | |||
| ctx, cancel := context.WithTimeout(ctx, m.dialTimeout) | |||
| defer cancel() | |||
| dialer := net.Dialer{} | |||
| tcpConn, err := dialer.DialContext(ctx, "tcp", fmt.Sprintf("%v:%v", endpoint.IP, endpoint.Port)) | |||
| if err != nil { | |||
| return nil, err | |||
| } | |||
| err = m.filterTCPConn(tcpConn) | |||
| if err != nil { | |||
| if err := tcpConn.Close(); err != nil { | |||
| m.logger.Debug("failed to close TCP connection", "err", err) | |||
| } | |||
| return nil, err | |||
| } | |||
| conn, err := newMConnConnection(m, tcpConn, endpoint.PeerID) | |||
| if err != nil { | |||
| m.conns.Remove(tcpConn) | |||
| if err := tcpConn.Close(); err != nil { | |||
| m.logger.Debug("failed to close TCP connection", "err", err) | |||
| } | |||
| return nil, err | |||
| } | |||
| return conn, nil | |||
| } | |||
| // Endpoints implements Transport. | |||
| func (m *MConnTransport) Endpoints() []Endpoint { | |||
| if m.listener == nil { | |||
| return []Endpoint{} | |||
| } | |||
| addr := m.listener.Addr().(*net.TCPAddr) | |||
| return []Endpoint{{ | |||
| Protocol: MConnProtocol, | |||
| PeerID: m.nodeInfo.ID(), | |||
| IP: addr.IP, | |||
| Port: uint16(addr.Port), | |||
| }} | |||
| } | |||
| // Close implements Transport. | |||
| func (m *MConnTransport) Close() error { | |||
| var err error | |||
| m.closeOnce.Do(func() { | |||
| // We have to close chClose first, so that accept() will detect | |||
| // the closure and not propagate the error. | |||
| close(m.chClose) | |||
| if m.listener != nil { | |||
| err = m.listener.Close() | |||
| } | |||
| }) | |||
| return err | |||
| } | |||
| // filterTCPConn filters a TCP connection, rejecting it if this function errors. | |||
| func (m *MConnTransport) filterTCPConn(tcpConn net.Conn) error { | |||
| if m.conns.Has(tcpConn) { | |||
| return ErrRejected{conn: tcpConn, isDuplicate: true} | |||
| } | |||
| host, _, err := net.SplitHostPort(tcpConn.RemoteAddr().String()) | |||
| if err != nil { | |||
| return err | |||
| } | |||
| ip := net.ParseIP(host) | |||
| if ip == nil { | |||
| return fmt.Errorf("connection address has invalid IP address %q", host) | |||
| } | |||
| // Apply filter callbacks. | |||
| chErr := make(chan error, len(m.connFilters)) | |||
| for _, connFilter := range m.connFilters { | |||
| go func(connFilter ConnFilterFunc) { | |||
| chErr <- connFilter(m.conns, tcpConn, []net.IP{ip}) | |||
| }(connFilter) | |||
| } | |||
| for i := 0; i < cap(chErr); i++ { | |||
| select { | |||
| case err := <-chErr: | |||
| if err != nil { | |||
| return ErrRejected{conn: tcpConn, err: err, isFiltered: true} | |||
| } | |||
| case <-time.After(m.filterTimeout): | |||
| return ErrFilterTimeout{} | |||
| } | |||
| } | |||
| // FIXME: Doesn't really make sense to set this here, but we preserve the | |||
| // behavior from the previous P2P transport implementation. This should | |||
| // be moved to the router. | |||
| m.conns.Set(tcpConn, []net.IP{ip}) | |||
| return nil | |||
| } | |||
| // normalizeEndpoint normalizes and validates an endpoint. | |||
| func (m *MConnTransport) normalizeEndpoint(endpoint *Endpoint) error { | |||
| if endpoint == nil { | |||
| return errors.New("nil endpoint") | |||
| } | |||
| if err := endpoint.Validate(); err != nil { | |||
| return err | |||
| } | |||
| if endpoint.Protocol == "" { | |||
| endpoint.Protocol = MConnProtocol | |||
| } | |||
| if endpoint.Protocol != MConnProtocol { | |||
| return fmt.Errorf("unsupported protocol %q", endpoint.Protocol) | |||
| } | |||
| if len(endpoint.IP) == 0 { | |||
| return errors.New("endpoint must have an IP address") | |||
| } | |||
| if endpoint.Path != "" { | |||
| return fmt.Errorf("endpoint cannot have path (got %q)", endpoint.Path) | |||
| } | |||
| if endpoint.Port == 0 { | |||
| endpoint.Port = 26657 | |||
| } | |||
| return nil | |||
| } | |||
| // mConnConnection implements Connection for MConnTransport. It takes a base TCP | |||
| // connection and upgrades it to MConnection over an encrypted SecretConnection. | |||
| type mConnConnection struct { | |||
| logger log.Logger | |||
| transport *MConnTransport | |||
| secretConn *conn.SecretConnection | |||
| mConn *conn.MConnection | |||
| peerInfo DefaultNodeInfo | |||
| closeOnce sync.Once | |||
| chReceive chan mConnMessage | |||
| chError chan error | |||
| chClose chan struct{} | |||
| } | |||
| // mConnMessage passes MConnection messages through internal channels. | |||
| type mConnMessage struct { | |||
| channelID byte | |||
| payload []byte | |||
| } | |||
| // newMConnConnection creates a new mConnConnection by handshaking | |||
| // with a peer. | |||
| func newMConnConnection( | |||
| transport *MConnTransport, | |||
| tcpConn net.Conn, | |||
| expectPeerID ID, | |||
| ) (c *mConnConnection, err error) { | |||
| // FIXME: Since the MConnection code panics, we need to recover here | |||
| // and turn it into an error. Be careful not to alias err, so we can | |||
| // update it from within this function. We should remove panics instead. | |||
| defer func() { | |||
| if r := recover(); r != nil { | |||
| err = ErrRejected{ | |||
| conn: tcpConn, | |||
| err: fmt.Errorf("recovered from panic: %v", r), | |||
| isAuthFailure: true, | |||
| } | |||
| } | |||
| }() | |||
| err = tcpConn.SetDeadline(time.Now().Add(transport.handshakeTimeout)) | |||
| if err != nil { | |||
| err = ErrRejected{ | |||
| conn: tcpConn, | |||
| err: fmt.Errorf("secret conn failed: %v", err), | |||
| isAuthFailure: true, | |||
| } | |||
| return | |||
| } | |||
| c = &mConnConnection{ | |||
| transport: transport, | |||
| chReceive: make(chan mConnMessage), | |||
| chError: make(chan error), | |||
| chClose: make(chan struct{}), | |||
| } | |||
| c.secretConn, err = conn.MakeSecretConnection(tcpConn, transport.privKey) | |||
| if err != nil { | |||
| err = ErrRejected{ | |||
| conn: tcpConn, | |||
| err: fmt.Errorf("secret conn failed: %v", err), | |||
| isAuthFailure: true, | |||
| } | |||
| return | |||
| } | |||
| c.peerInfo, err = c.handshake() | |||
| if err != nil { | |||
| err = ErrRejected{ | |||
| conn: tcpConn, | |||
| err: fmt.Errorf("handshake failed: %v", err), | |||
| isAuthFailure: true, | |||
| } | |||
| return | |||
| } | |||
| // Validate node info. | |||
| // FIXME: All of the ID verification code below should be moved to the | |||
| // router once implemented. | |||
| err = c.peerInfo.Validate() | |||
| if err != nil { | |||
| err = ErrRejected{ | |||
| conn: tcpConn, | |||
| err: err, | |||
| isNodeInfoInvalid: true, | |||
| } | |||
| return | |||
| } | |||
| // For outgoing conns, ensure connection key matches dialed key. | |||
| if expectPeerID != "" { | |||
| peerID := PubKeyToID(c.PubKey()) | |||
| if expectPeerID != peerID { | |||
| err = ErrRejected{ | |||
| conn: tcpConn, | |||
| id: peerID, | |||
| err: fmt.Errorf( | |||
| "conn.ID (%v) dialed ID (%v) mismatch", | |||
| peerID, | |||
| expectPeerID, | |||
| ), | |||
| isAuthFailure: true, | |||
| } | |||
| return | |||
| } | |||
| } | |||
| // Reject self. | |||
| if transport.nodeInfo.ID() == c.peerInfo.ID() { | |||
| err = ErrRejected{ | |||
| addr: *NewNetAddress(c.peerInfo.ID(), c.secretConn.RemoteAddr()), | |||
| conn: tcpConn, | |||
| id: c.peerInfo.ID(), | |||
| isSelf: true, | |||
| } | |||
| return | |||
| } | |||
| err = transport.nodeInfo.CompatibleWith(c.peerInfo) | |||
| if err != nil { | |||
| err = ErrRejected{ | |||
| conn: tcpConn, | |||
| err: err, | |||
| id: c.peerInfo.ID(), | |||
| isIncompatible: true, | |||
| } | |||
| return | |||
| } | |||
| err = tcpConn.SetDeadline(time.Time{}) | |||
| if err != nil { | |||
| err = ErrRejected{ | |||
| conn: tcpConn, | |||
| err: fmt.Errorf("secret conn failed: %v", err), | |||
| isAuthFailure: true, | |||
| } | |||
| return | |||
| } | |||
| // Set up the MConnection wrapper | |||
| c.mConn = conn.NewMConnectionWithConfig( | |||
| c.secretConn, | |||
| transport.channelDescs, | |||
| c.onReceive, | |||
| c.onError, | |||
| transport.mConnConfig, | |||
| ) | |||
| // FIXME: Log format is set up for compatibility with existing peer code. | |||
| c.logger = transport.logger.With("peer", c.RemoteEndpoint().NetAddress()) | |||
| c.mConn.SetLogger(c.logger) | |||
| err = c.mConn.Start() | |||
| return c, err | |||
| } | |||
| // handshake performs an MConn handshake, returning the peer's node info. | |||
| func (c *mConnConnection) handshake() (DefaultNodeInfo, error) { | |||
| var pbNodeInfo p2pproto.DefaultNodeInfo | |||
| chErr := make(chan error, 2) | |||
| go func() { | |||
| _, err := protoio.NewDelimitedWriter(c.secretConn).WriteMsg(c.transport.nodeInfo.ToProto()) | |||
| chErr <- err | |||
| }() | |||
| go func() { | |||
| chErr <- protoio.NewDelimitedReader(c.secretConn, MaxNodeInfoSize()).ReadMsg(&pbNodeInfo) | |||
| }() | |||
| for i := 0; i < cap(chErr); i++ { | |||
| if err := <-chErr; err != nil { | |||
| return DefaultNodeInfo{}, err | |||
| } | |||
| } | |||
| return DefaultNodeInfoFromProto(&pbNodeInfo) | |||
| } | |||
| // onReceive is a callback for MConnection received messages. | |||
| func (c *mConnConnection) onReceive(channelID byte, payload []byte) { | |||
| select { | |||
| case c.chReceive <- mConnMessage{channelID: channelID, payload: payload}: | |||
| case <-c.chClose: | |||
| } | |||
| } | |||
| // onError is a callback for MConnection errors. The error is passed to | |||
| // chError, which is only consumed by ReceiveMessage() for parity with | |||
| // the old MConnection behavior. | |||
| func (c *mConnConnection) onError(e interface{}) { | |||
| err, ok := e.(error) | |||
| if !ok { | |||
| err = fmt.Errorf("%v", err) | |||
| } | |||
| select { | |||
| case c.chError <- err: | |||
| case <-c.chClose: | |||
| } | |||
| } | |||
| // String displays connection information. | |||
| // FIXME: This is here for backwards compatibility with existing code, | |||
| // it should probably just return RemoteEndpoint().String(), if anything. | |||
| func (c *mConnConnection) String() string { | |||
| endpoint := c.RemoteEndpoint() | |||
| return fmt.Sprintf("MConn{%v:%v}", endpoint.IP, endpoint.Port) | |||
| } | |||
| // SendMessage implements Connection. | |||
| func (c *mConnConnection) SendMessage(channelID byte, msg []byte) (bool, error) { | |||
| // We don't check chError here, to preserve old MConnection behavior. | |||
| select { | |||
| case <-c.chClose: | |||
| return false, io.EOF | |||
| default: | |||
| return c.mConn.Send(channelID, msg), nil | |||
| } | |||
| } | |||
| // TrySendMessage implements Connection. | |||
| func (c *mConnConnection) TrySendMessage(channelID byte, msg []byte) (bool, error) { | |||
| // We don't check chError here, to preserve old MConnection behavior. | |||
| select { | |||
| case <-c.chClose: | |||
| return false, io.EOF | |||
| default: | |||
| return c.mConn.TrySend(channelID, msg), nil | |||
| } | |||
| } | |||
| // ReceiveMessage implements Connection. | |||
| func (c *mConnConnection) ReceiveMessage() (byte, []byte, error) { | |||
| select { | |||
| case err := <-c.chError: | |||
| return 0, nil, err | |||
| case <-c.chClose: | |||
| return 0, nil, io.EOF | |||
| case msg := <-c.chReceive: | |||
| return msg.channelID, msg.payload, nil | |||
| } | |||
| } | |||
| // NodeInfo implements Connection. | |||
| func (c *mConnConnection) NodeInfo() DefaultNodeInfo { | |||
| return c.peerInfo | |||
| } | |||
| // PubKey implements Connection. | |||
| func (c *mConnConnection) PubKey() crypto.PubKey { | |||
| return c.secretConn.RemotePubKey() | |||
| } | |||
| // LocalEndpoint implements Connection. | |||
| func (c *mConnConnection) LocalEndpoint() Endpoint { | |||
| // FIXME: For compatibility with existing P2P tests we need to | |||
| // handle non-TCP connections. This should be removed. | |||
| endpoint := Endpoint{ | |||
| Protocol: MConnProtocol, | |||
| PeerID: c.transport.nodeInfo.ID(), | |||
| } | |||
| if addr, ok := c.secretConn.LocalAddr().(*net.TCPAddr); ok { | |||
| endpoint.IP = addr.IP | |||
| endpoint.Port = uint16(addr.Port) | |||
| } | |||
| return endpoint | |||
| } | |||
| // RemoteEndpoint implements Connection. | |||
| func (c *mConnConnection) RemoteEndpoint() Endpoint { | |||
| // FIXME: For compatibility with existing P2P tests we need to | |||
| // handle non-TCP connections. This should be removed. | |||
| endpoint := Endpoint{ | |||
| Protocol: MConnProtocol, | |||
| PeerID: c.peerInfo.ID(), | |||
| } | |||
| if addr, ok := c.secretConn.RemoteAddr().(*net.TCPAddr); ok { | |||
| endpoint.IP = addr.IP | |||
| endpoint.Port = uint16(addr.Port) | |||
| } | |||
| return endpoint | |||
| } | |||
| // Status implements Connection. | |||
| func (c *mConnConnection) Status() conn.ConnectionStatus { | |||
| return c.mConn.Status() | |||
| } | |||
| // Close implements Connection. | |||
| func (c *mConnConnection) Close() error { | |||
| c.transport.conns.RemoveAddr(c.secretConn.RemoteAddr()) | |||
| var err error | |||
| c.closeOnce.Do(func() { | |||
| err = c.mConn.Stop() | |||
| close(c.chClose) | |||
| }) | |||
| return err | |||
| } | |||
| // FlushClose implements Connection. | |||
| func (c *mConnConnection) FlushClose() error { | |||
| c.transport.conns.RemoveAddr(c.secretConn.RemoteAddr()) | |||
| c.closeOnce.Do(func() { | |||
| c.mConn.FlushStop() | |||
| close(c.chClose) | |||
| }) | |||
| return nil | |||
| } | |||
+ 0
- 696
p2p/transport_test.go
View File
| @ -1,696 +0,0 @@ | |||
| package p2p | |||
| import ( | |||
| "fmt" | |||
| "math/rand" | |||
| "net" | |||
| "reflect" | |||
| "runtime" | |||
| "strings" | |||
| "testing" | |||
| "time" | |||
| "github.com/tendermint/tendermint/crypto/ed25519" | |||
| "github.com/tendermint/tendermint/libs/protoio" | |||
| "github.com/tendermint/tendermint/p2p/conn" | |||
| tmp2p "github.com/tendermint/tendermint/proto/tendermint/p2p" | |||
| ) | |||
| var defaultNodeName = "host_peer" | |||
| func emptyNodeInfo() NodeInfo { | |||
| return DefaultNodeInfo{} | |||
| } | |||
| // newMultiplexTransport returns a tcp connected multiplexed peer | |||
| // using the default MConnConfig. It's a convenience function used | |||
| // for testing. | |||
| func newMultiplexTransport( | |||
| nodeInfo NodeInfo, | |||
| nodeKey NodeKey, | |||
| ) *MultiplexTransport { | |||
| return NewMultiplexTransport( | |||
| nodeInfo, nodeKey, conn.DefaultMConnConfig(), | |||
| ) | |||
| } | |||
| func TestTransportMultiplexConnFilter(t *testing.T) { | |||
| mt := newMultiplexTransport( | |||
| emptyNodeInfo(), | |||
| GenNodeKey(), | |||
| ) | |||
| id := mt.nodeKey.ID | |||
| MultiplexTransportConnFilters( | |||
| func(_ ConnSet, _ net.Conn, _ []net.IP) error { return nil }, | |||
| func(_ ConnSet, _ net.Conn, _ []net.IP) error { return nil }, | |||
| func(_ ConnSet, _ net.Conn, _ []net.IP) error { | |||
| return fmt.Errorf("rejected") | |||
| }, | |||
| )(mt) | |||
| addr, err := NewNetAddressString(IDAddressString(id, "127.0.0.1:0")) | |||
| if err != nil { | |||
| t.Fatal(err) | |||
| } | |||
| if err := mt.Listen(*addr); err != nil { | |||
| t.Fatal(err) | |||
| } | |||
| errc := make(chan error) | |||
| go func() { | |||
| addr := NewNetAddress(id, mt.listener.Addr()) | |||
| _, err := addr.Dial() | |||
| if err != nil { | |||
| errc <- err | |||
| return | |||
| } | |||
| close(errc) | |||
| }() | |||
| if err := <-errc; err != nil { | |||
| t.Errorf("connection failed: %v", err) | |||
| } | |||
| _, err = mt.Accept(peerConfig{}) | |||
| if err, ok := err.(ErrRejected); ok { | |||
| if !err.IsFiltered() { | |||
| t.Errorf("expected peer to be filtered, got %v", err) | |||
| } | |||
| } else { | |||
| t.Errorf("expected ErrRejected, got %v", err) | |||
| } | |||
| } | |||
| func TestTransportMultiplexConnFilterTimeout(t *testing.T) { | |||
| mt := newMultiplexTransport( | |||
| emptyNodeInfo(), | |||
| GenNodeKey(), | |||
| ) | |||
| id := mt.nodeKey.ID | |||
| MultiplexTransportFilterTimeout(5 * time.Millisecond)(mt) | |||
| MultiplexTransportConnFilters( | |||
| func(_ ConnSet, _ net.Conn, _ []net.IP) error { | |||
| time.Sleep(1 * time.Second) | |||
| return nil | |||
| }, | |||
| )(mt) | |||
| addr, err := NewNetAddressString(IDAddressString(id, "127.0.0.1:0")) | |||
| if err != nil { | |||
| t.Fatal(err) | |||
| } | |||
| if err := mt.Listen(*addr); err != nil { | |||
| t.Fatal(err) | |||
| } | |||
| errc := make(chan error) | |||
| go func() { | |||
| addr := NewNetAddress(id, mt.listener.Addr()) | |||
| _, err := addr.Dial() | |||
| if err != nil { | |||
| errc <- err | |||
| return | |||
| } | |||
| close(errc) | |||
| }() | |||
| if err := <-errc; err != nil { | |||
| t.Errorf("connection failed: %v", err) | |||
| } | |||
| _, err = mt.Accept(peerConfig{}) | |||
| if _, ok := err.(ErrFilterTimeout); !ok { | |||
| t.Errorf("expected ErrFilterTimeout, got %v", err) | |||
| } | |||
| } | |||
| func TestTransportMultiplexMaxIncomingConnections(t *testing.T) { | |||
| pv := ed25519.GenPrivKey() | |||
| id := PubKeyToID(pv.PubKey()) | |||
| mt := newMultiplexTransport( | |||
| testNodeInfo( | |||
| id, "transport", | |||
| ), | |||
| NodeKey{ | |||
| ID: id, | |||
| PrivKey: pv, | |||
| }, | |||
| ) | |||
| MultiplexTransportMaxIncomingConnections(0)(mt) | |||
| addr, err := NewNetAddressString(IDAddressString(id, "127.0.0.1:0")) | |||
| if err != nil { | |||
| t.Fatal(err) | |||
| } | |||
| const maxIncomingConns = 2 | |||
| MultiplexTransportMaxIncomingConnections(maxIncomingConns)(mt) | |||
| if err := mt.Listen(*addr); err != nil { | |||
| t.Fatal(err) | |||
| } | |||
| laddr := NewNetAddress(mt.nodeKey.ID, mt.listener.Addr()) | |||
| // Connect more peers than max | |||
| for i := 0; i <= maxIncomingConns; i++ { | |||
| errc := make(chan error) | |||
| go testDialer(*laddr, errc) | |||
| err = <-errc | |||
| if i < maxIncomingConns { | |||
| if err != nil { | |||
| t.Errorf("dialer connection failed: %v", err) | |||
| } | |||
| _, err = mt.Accept(peerConfig{}) | |||
| if err != nil { | |||
| t.Errorf("connection failed: %v", err) | |||
| } | |||
| } else if err == nil || !strings.Contains(err.Error(), "i/o timeout") { | |||
| // mt actually blocks forever on trying to accept a new peer into a full channel so | |||
| // expect the dialer to encounter a timeout error. Calling mt.Accept will block until | |||
| // mt is closed. | |||
| t.Errorf("expected i/o timeout error, got %v", err) | |||
| } | |||
| } | |||
| } | |||
| func TestTransportMultiplexAcceptMultiple(t *testing.T) { | |||
| mt := testSetupMultiplexTransport(t) | |||
| laddr := NewNetAddress(mt.nodeKey.ID, mt.listener.Addr()) | |||
| var ( | |||
| seed = rand.New(rand.NewSource(time.Now().UnixNano())) | |||
| nDialers = seed.Intn(64) + 64 | |||
| errc = make(chan error, nDialers) | |||
| ) | |||
| // Setup dialers. | |||
| for i := 0; i < nDialers; i++ { | |||
| go testDialer(*laddr, errc) | |||
| } | |||
| // Catch connection errors. | |||
| for i := 0; i < nDialers; i++ { | |||
| if err := <-errc; err != nil { | |||
| t.Fatal(err) | |||
| } | |||
| } | |||
| ps := []Peer{} | |||
| // Accept all peers. | |||
| for i := 0; i < cap(errc); i++ { | |||
| p, err := mt.Accept(peerConfig{}) | |||
| if err != nil { | |||
| t.Fatal(err) | |||
| } | |||
| if err := p.Start(); err != nil { | |||
| t.Fatal(err) | |||
| } | |||
| ps = append(ps, p) | |||
| } | |||
| if have, want := len(ps), cap(errc); have != want { | |||
| t.Errorf("have %v, want %v", have, want) | |||
| } | |||
| // Stop all peers. | |||
| for _, p := range ps { | |||
| if err := p.Stop(); err != nil { | |||
| t.Fatal(err) | |||
| } | |||
| } | |||
| if err := mt.Close(); err != nil { | |||
| t.Errorf("close errored: %v", err) | |||
| } | |||
| } | |||
| func testDialer(dialAddr NetAddress, errc chan error) { | |||
| var ( | |||
| pv = ed25519.GenPrivKey() | |||
| id = PubKeyToID(pv.PubKey()) | |||
| dialer = newMultiplexTransport( | |||
| testNodeInfo(id, defaultNodeName), | |||
| NodeKey{ | |||
| ID: id, | |||
| PrivKey: pv, | |||
| }, | |||
| ) | |||
| ) | |||
| _, err := dialer.Dial(dialAddr, peerConfig{}) | |||
| if err != nil { | |||
| errc <- err | |||
| return | |||
| } | |||
| // Signal that the connection was established. | |||
| errc <- nil | |||
| } | |||
| func TestTransportMultiplexAcceptNonBlocking(t *testing.T) { | |||
| mt := testSetupMultiplexTransport(t) | |||
| var ( | |||
| fastNodePV = ed25519.GenPrivKey() | |||
| fastNodeInfo = testNodeInfo(PubKeyToID(fastNodePV.PubKey()), "fastnode") | |||
| errc = make(chan error) | |||
| fastc = make(chan struct{}) | |||
| slowc = make(chan struct{}) | |||
| slowdonec = make(chan struct{}) | |||
| ) | |||
| // Simulate slow Peer. | |||
| go func() { | |||
| addr := NewNetAddress(mt.nodeKey.ID, mt.listener.Addr()) | |||
| c, err := addr.Dial() | |||
| if err != nil { | |||
| errc <- err | |||
| return | |||
| } | |||
| close(slowc) | |||
| defer func() { | |||
| close(slowdonec) | |||
| }() | |||
| // Make sure we switch to fast peer goroutine. | |||
| runtime.Gosched() | |||
| select { | |||
| case <-fastc: | |||
| // Fast peer connected. | |||
| case <-time.After(200 * time.Millisecond): | |||
| // We error if the fast peer didn't succeed. | |||
| errc <- fmt.Errorf("fast peer timed out") | |||
| } | |||
| sc, err := upgradeSecretConn(c, 200*time.Millisecond, ed25519.GenPrivKey()) | |||
| if err != nil { | |||
| errc <- err | |||
| return | |||
| } | |||
| _, err = handshake(sc, 200*time.Millisecond, | |||
| testNodeInfo( | |||
| PubKeyToID(ed25519.GenPrivKey().PubKey()), | |||
| "slow_peer", | |||
| )) | |||
| if err != nil { | |||
| errc <- err | |||
| } | |||
| }() | |||
| // Simulate fast Peer. | |||
| go func() { | |||
| <-slowc | |||
| var ( | |||
| dialer = newMultiplexTransport( | |||
| fastNodeInfo, | |||
| NodeKey{ | |||
| ID: PubKeyToID(fastNodePV.PubKey()), | |||
| PrivKey: fastNodePV, | |||
| }, | |||
| ) | |||
| ) | |||
| addr := NewNetAddress(mt.nodeKey.ID, mt.listener.Addr()) | |||
| _, err := dialer.Dial(*addr, peerConfig{}) | |||
| if err != nil { | |||
| errc <- err | |||
| return | |||
| } | |||
| close(fastc) | |||
| <-slowdonec | |||
| close(errc) | |||
| }() | |||
| if err := <-errc; err != nil { | |||
| t.Logf("connection failed: %v", err) | |||
| } | |||
| p, err := mt.Accept(peerConfig{}) | |||
| if err != nil { | |||
| t.Fatal(err) | |||
| } | |||
| if have, want := p.NodeInfo(), fastNodeInfo; !reflect.DeepEqual(have, want) { | |||
| t.Errorf("have %v, want %v", have, want) | |||
| } | |||
| } | |||
| func TestTransportMultiplexValidateNodeInfo(t *testing.T) { | |||
| mt := testSetupMultiplexTransport(t) | |||
| errc := make(chan error) | |||
| go func() { | |||
| var ( | |||
| pv = ed25519.GenPrivKey() | |||
| id = PubKeyToID(pv.PubKey()) | |||
| dialer = newMultiplexTransport( | |||
| testNodeInfo(id, ""), // Should not be empty | |||
| NodeKey{ | |||
| ID: id, | |||
| PrivKey: pv, | |||
| }, | |||
| ) | |||
| ) | |||
| addr := NewNetAddress(mt.nodeKey.ID, mt.listener.Addr()) | |||
| _, err := dialer.Dial(*addr, peerConfig{}) | |||
| if err != nil { | |||
| errc <- err | |||
| return | |||
| } | |||
| close(errc) | |||
| }() | |||
| if err := <-errc; err != nil { | |||
| t.Errorf("connection failed: %v", err) | |||
| } | |||
| _, err := mt.Accept(peerConfig{}) | |||
| if err, ok := err.(ErrRejected); ok { | |||
| if !err.IsNodeInfoInvalid() { | |||
| t.Errorf("expected NodeInfo to be invalid, got %v", err) | |||
| } | |||
| } else { | |||
| t.Errorf("expected ErrRejected, got %v", err) | |||
| } | |||
| } | |||
| func TestTransportMultiplexRejectMissmatchID(t *testing.T) { | |||
| mt := testSetupMultiplexTransport(t) | |||
| errc := make(chan error) | |||
| go func() { | |||
| dialer := newMultiplexTransport( | |||
| testNodeInfo( | |||
| PubKeyToID(ed25519.GenPrivKey().PubKey()), "dialer", | |||
| ), | |||
| GenNodeKey(), | |||
| ) | |||
| addr := NewNetAddress(mt.nodeKey.ID, mt.listener.Addr()) | |||
| _, err := dialer.Dial(*addr, peerConfig{}) | |||
| if err != nil { | |||
| errc <- err | |||
| return | |||
| } | |||
| close(errc) | |||
| }() | |||
| if err := <-errc; err != nil { | |||
| t.Errorf("connection failed: %v", err) | |||
| } | |||
| _, err := mt.Accept(peerConfig{}) | |||
| if err, ok := err.(ErrRejected); ok { | |||
| if !err.IsAuthFailure() { | |||
| t.Errorf("expected auth failure, got %v", err) | |||
| } | |||
| } else { | |||
| t.Errorf("expected ErrRejected, got %v", err) | |||
| } | |||
| } | |||
| func TestTransportMultiplexDialRejectWrongID(t *testing.T) { | |||
| mt := testSetupMultiplexTransport(t) | |||
| var ( | |||
| pv = ed25519.GenPrivKey() | |||
| id = PubKeyToID(pv.PubKey()) | |||
| dialer = newMultiplexTransport( | |||
| testNodeInfo(id, ""), // Should not be empty | |||
| NodeKey{ | |||
| ID: id, | |||
| PrivKey: pv, | |||
| }, | |||
| ) | |||
| ) | |||
| wrongID := PubKeyToID(ed25519.GenPrivKey().PubKey()) | |||
| addr := NewNetAddress(wrongID, mt.listener.Addr()) | |||
| _, err := dialer.Dial(*addr, peerConfig{}) | |||
| if err != nil { | |||
| t.Logf("connection failed: %v", err) | |||
| if err, ok := err.(ErrRejected); ok { | |||
| if !err.IsAuthFailure() { | |||
| t.Errorf("expected auth failure, got %v", err) | |||
| } | |||
| } else { | |||
| t.Errorf("expected ErrRejected, got %v", err) | |||
| } | |||
| } | |||
| } | |||
| func TestTransportMultiplexRejectIncompatible(t *testing.T) { | |||
| mt := testSetupMultiplexTransport(t) | |||
| errc := make(chan error) | |||
| go func() { | |||
| var ( | |||
| pv = ed25519.GenPrivKey() | |||
| id = PubKeyToID(pv.PubKey()) | |||
| dialer = newMultiplexTransport( | |||
| testNodeInfoWithNetwork(id, "dialer", "incompatible-network"), | |||
| NodeKey{ | |||
| ID: id, | |||
| PrivKey: pv, | |||
| }, | |||
| ) | |||
| ) | |||
| addr := NewNetAddress(mt.nodeKey.ID, mt.listener.Addr()) | |||
| _, err := dialer.Dial(*addr, peerConfig{}) | |||
| if err != nil { | |||
| errc <- err | |||
| return | |||
| } | |||
| close(errc) | |||
| }() | |||
| _, err := mt.Accept(peerConfig{}) | |||
| if err, ok := err.(ErrRejected); ok { | |||
| if !err.IsIncompatible() { | |||
| t.Errorf("expected to reject incompatible, got %v", err) | |||
| } | |||
| } else { | |||
| t.Errorf("expected ErrRejected, got %v", err) | |||
| } | |||
| } | |||
| func TestTransportMultiplexRejectSelf(t *testing.T) { | |||
| mt := testSetupMultiplexTransport(t) | |||
| errc := make(chan error) | |||
| go func() { | |||
| addr := NewNetAddress(mt.nodeKey.ID, mt.listener.Addr()) | |||
| _, err := mt.Dial(*addr, peerConfig{}) | |||
| if err != nil { | |||
| errc <- err | |||
| return | |||
| } | |||
| close(errc) | |||
| }() | |||
| if err := <-errc; err != nil { | |||
| if err, ok := err.(ErrRejected); ok { | |||
| if !err.IsSelf() { | |||
| t.Errorf("expected to reject self, got: %v", err) | |||
| } | |||
| } else { | |||
| t.Errorf("expected ErrRejected, got %v", err) | |||
| } | |||
| } else { | |||
| t.Errorf("expected connection failure") | |||
| } | |||
| _, err := mt.Accept(peerConfig{}) | |||
| if err, ok := err.(ErrRejected); ok { | |||
| if !err.IsSelf() { | |||
| t.Errorf("expected to reject self, got: %v", err) | |||
| } | |||
| } else { | |||
| t.Errorf("expected ErrRejected, got %v", nil) | |||
| } | |||
| } | |||
| func TestTransportConnDuplicateIPFilter(t *testing.T) { | |||
| filter := ConnDuplicateIPFilter() | |||
| if err := filter(nil, &testTransportConn{}, nil); err != nil { | |||
| t.Fatal(err) | |||
| } | |||
| var ( | |||
| c = &testTransportConn{} | |||
| cs = NewConnSet() | |||
| ) | |||
| cs.Set(c, []net.IP{ | |||
| {10, 0, 10, 1}, | |||
| {10, 0, 10, 2}, | |||
| {10, 0, 10, 3}, | |||
| }) | |||
| if err := filter(cs, c, []net.IP{ | |||
| {10, 0, 10, 2}, | |||
| }); err == nil { | |||
| t.Errorf("expected Peer to be rejected as duplicate") | |||
| } | |||
| } | |||
| func TestTransportHandshake(t *testing.T) { | |||
| ln, err := net.Listen("tcp", "127.0.0.1:0") | |||
| if err != nil { | |||
| t.Fatal(err) | |||
| } | |||
| var ( | |||
| peerPV = ed25519.GenPrivKey() | |||
| peerNodeInfo = testNodeInfo(PubKeyToID(peerPV.PubKey()), defaultNodeName) | |||
| ) | |||
| go func() { | |||
| c, err := net.Dial(ln.Addr().Network(), ln.Addr().String()) | |||
| if err != nil { | |||
| t.Error(err) | |||
| return | |||
| } | |||
| go func(c net.Conn) { | |||
| _, err := protoio.NewDelimitedWriter(c).WriteMsg(peerNodeInfo.(DefaultNodeInfo).ToProto()) | |||
| if err != nil { | |||
| t.Error(err) | |||
| } | |||
| }(c) | |||
| go func(c net.Conn) { | |||
| var ( | |||
| // ni DefaultNodeInfo | |||
| pbni tmp2p.DefaultNodeInfo | |||
| ) | |||
| protoReader := protoio.NewDelimitedReader(c, MaxNodeInfoSize()) | |||
| err := protoReader.ReadMsg(&pbni) | |||
| if err != nil { | |||
| t.Error(err) | |||
| } | |||
| _, err = DefaultNodeInfoFromToProto(&pbni) | |||
| if err != nil { | |||
| t.Error(err) | |||
| } | |||
| }(c) | |||
| }() | |||
| c, err := ln.Accept() | |||
| if err != nil { | |||
| t.Fatal(err) | |||
| } | |||
| ni, err := handshake(c, 20*time.Millisecond, emptyNodeInfo()) | |||
| if err != nil { | |||
| t.Fatal(err) | |||
| } | |||
| if have, want := ni, peerNodeInfo; !reflect.DeepEqual(have, want) { | |||
| t.Errorf("have %v, want %v", have, want) | |||
| } | |||
| } | |||
| // create listener | |||
| func testSetupMultiplexTransport(t *testing.T) *MultiplexTransport { | |||
| var ( | |||
| pv = ed25519.GenPrivKey() | |||
| id = PubKeyToID(pv.PubKey()) | |||
| mt = newMultiplexTransport( | |||
| testNodeInfo( | |||
| id, "transport", | |||
| ), | |||
| NodeKey{ | |||
| ID: id, | |||
| PrivKey: pv, | |||
| }, | |||
| ) | |||
| ) | |||
| addr, err := NewNetAddressString(IDAddressString(id, "127.0.0.1:0")) | |||
| if err != nil { | |||
| t.Fatal(err) | |||
| } | |||
| if err := mt.Listen(*addr); err != nil { | |||
| t.Fatal(err) | |||
| } | |||
| // give the listener some time to get ready | |||
| time.Sleep(20 * time.Millisecond) | |||
| return mt | |||
| } | |||
| type testTransportAddr struct{} | |||
| func (a *testTransportAddr) Network() string { return "tcp" } | |||
| func (a *testTransportAddr) String() string { return "test.local:1234" } | |||
| type testTransportConn struct{} | |||
| func (c *testTransportConn) Close() error { | |||
| return fmt.Errorf("close() not implemented") | |||
| } | |||
| func (c *testTransportConn) LocalAddr() net.Addr { | |||
| return &testTransportAddr{} | |||
| } | |||
| func (c *testTransportConn) RemoteAddr() net.Addr { | |||
| return &testTransportAddr{} | |||
| } | |||
| func (c *testTransportConn) Read(_ []byte) (int, error) { | |||
| return -1, fmt.Errorf("read() not implemented") | |||
| } | |||
| func (c *testTransportConn) SetDeadline(_ time.Time) error { | |||
| return fmt.Errorf("setDeadline() not implemented") | |||
| } | |||
| func (c *testTransportConn) SetReadDeadline(_ time.Time) error { | |||
| return fmt.Errorf("setReadDeadline() not implemented") | |||
| } | |||
| func (c *testTransportConn) SetWriteDeadline(_ time.Time) error { | |||
| return fmt.Errorf("setWriteDeadline() not implemented") | |||
| } | |||
| func (c *testTransportConn) Write(_ []byte) (int, error) { | |||
| return -1, fmt.Errorf("write() not implemented") | |||
| } | |||