Repository: pion/dtls
Pull request #184 (2020-02-03 04:51:47)

Make Conn fully net.Conn compatible
* Implement `Read/WriteDeadline`
* Return remained data on `Read` after `Close`
* Move network related internal packages under `internal/net/`
* Test using `nettest.TestConn`

### Reference issue
related: #148

at-wat: passed `nettest.TestConn`
codecov-io: # [Codecov](https://codecov.io/gh/pion/dtls/pull/184?src=pr&el=h1) Report
> Merging [#184](https://codecov.io/gh/pion/dtls/pull/184?src=pr&el=desc) into [master](https://codecov.io/gh/pion/dtls/commit/0b80bd9abfa64f92945b81c8acd9407d741a50ed?src=pr&el=desc) will **increase** coverage by `1.42%`.
> The diff coverage is `91.11%`.

[![Impacted file tree graph](https://codecov.io/gh/pion/dtls/pull/184/graphs/tree.svg?width=650&token=bcCB2jB88I&height=150&src=pr)](https://codecov.io/gh/pion/dtls/pull/184?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master     #184      +/-   ##
==========================================
+ Coverage    72.6%   74.02%   +1.42%     
==========================================
  Files          60       61       +1     
  Lines        3431     3523      +92     
==========================================
+ Hits         2491     2608     +117     
+ Misses        649      629      -20     
+ Partials      291      286       -5
```

| Flag | Coverage Δ | |
|---|---|---|
| #go | `74.02% <91.11%> (+1.42%)` | :arrow_up: |
| #wasm | `?` | |

| [Impacted Files](https://codecov.io/gh/pion/dtls/pull/184?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [internal/net/deadline/deadline.go](https://codecov.io/gh/pion/dtls/pull/184/diff?src=pr&el=tree#diff-aW50ZXJuYWwvbmV0L2RlYWRsaW5lL2RlYWRsaW5lLmdv) | `100% <100%> (ø)` | |
| [conn.go](https://codecov.io/gh/pion/dtls/pull/184/diff?src=pr&el=tree#diff-Y29ubi5nbw==) | `81.85% <77.41%> (+4.35%)` | :arrow_up: |
| [listener.go](https://codecov.io/gh/pion/dtls/pull/184/diff?src=pr&el=tree#diff-bGlzdGVuZXIuZ28=) | `57.89% <83.33%> (ø)` | :arrow_up: |
| [internal/net/udp/conn.go](https://codecov.io/gh/pion/dtls/pull/184/diff?src=pr&el=tree#diff-aW50ZXJuYWwvbmV0L3VkcC9jb25uLmdv) | `81.96% <85.71%> (ø)` | |
| [internal/net/dpipe/dpipe.go](https://codecov.io/gh/pion/dtls/pull/184/diff?src=pr&el=tree#diff-aW50ZXJuYWwvbmV0L2RwaXBlL2RwaXBlLmdv) | `96.47% <96.47%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/pion/dtls/pull/184?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/pion/dtls/pull/184?src=pr&el=footer). Last update [0b80bd9...4fd3cb7](https://codecov.io/gh/pion/dtls/pull/184?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

at-wat: I set branch of the dtls module to this at https://github.com/pion/webrtc/pull/1004. All tests seem working correctly.
daenney: Ah excellent. :shipit:
at-wat: rebased

Patch:
diff --git a/conn.go b/conn.go
index e38a1c8..68690b7 100644
--- a/conn.go
+++ b/conn.go
@@ -1,6 +1,7 @@
 package dtls
 
 import (
+	"context"
 	"crypto/rand"
 	"crypto/tls"
 	"crypto/x509"
@@ -13,6 +14,7 @@ import (
 	"time"
 
 	"github.com/pion/dtls/v2/internal/closer"
+	"github.com/pion/dtls/v2/internal/net/deadline"
 	"github.com/pion/logging"
 )
 
@@ -88,6 +90,9 @@ type Conn struct {
 	handshakeErr     *atomicError   // Error if one occurred during handshake
 	readErr          *atomicError   // Error if one occurred in inboundLoop
 
+	readDeadline  *deadline.Deadline
+	writeDeadline *deadline.Deadline
+
 	log logging.LeveledLogger
 }
 
@@ -166,6 +171,9 @@ func createConn(nextConn net.Conn, flightHandler flightHandler, handshakeMessage
 		log:                 logger,
 		handshakeErr:        &atomicError{},
 		readErr:             &atomicError{},
+
+		readDeadline:  deadline.New(),
+		writeDeadline: deadline.New(),
 	}
 
 	// Use host from conn address when serverName is not provided
@@ -255,33 +263,26 @@ func Server(conn net.Conn, config *Config) (*Conn, error) {
 
 // Read reads data from the connection.
 func (c *Conn) Read(p []byte) (n int, err error) {
-	checkConnStatus := func() error {
-		if err := c.handshakeErr.load(); err != nil {
-			return err
-		}
-		if c.connectionClosed.Err() != nil {
-			return io.EOF
-		}
-		if err := c.readErr.load(); err != nil {
-			return err
+	var out []byte
+	var ok bool
+	select {
+	case out, ok = <-c.decrypted:
+		// inboundLoop has closed but error has not been set yet
+		if !ok {
+			if err := c.handshakeErr.load(); err != nil {
+				return 0, err
+			}
+			if c.connectionClosed.Err() != nil {
+				return 0, io.EOF
+			}
+			if err := c.readErr.load(); err != nil {
+				return 0, err
+			}
+			return 0, io.EOF
 		}
-
-		return nil
-	}
-
-	if err := checkConnStatus(); err != nil {
-		return 0, err
-	}
-	out, ok := <-c.decrypted
-	if err := checkConnStatus(); err != nil {
-		return 0, err
+	case <-c.readDeadline.Done():
+		return 0, context.DeadlineExceeded
 	}
-
-	// inboundLoop has closed but error has not been set yet
-	if !ok {
-		return 0, io.EOF
-	}
-
 	if len(p) < len(out) {
 		return 0, errBufferTooSmall
 	}
@@ -291,8 +292,11 @@ func (c *Conn) Read(p []byte) (n int, err error) {
 
 // Write writes len(p) bytes from p to the DTLS connection
 func (c *Conn) Write(p []byte) (int, error) {
-	c.lock.Lock()
-	defer c.lock.Unlock()
+	select {
+	case <-c.writeDeadline.Done():
+		return 0, context.DeadlineExceeded
+	default:
+	}
 
 	if err := c.handshakeErr.load(); err != nil {
 		return 0, err
@@ -304,6 +308,9 @@ func (c *Conn) Write(p []byte) (int, error) {
 		return 0, errHandshakeInProgress
 	}
 
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
 	if err := c.bufferPacket(&packet{
 		record: &recordLayer{
 			recordLayerHeader: recordLayerHeader{
@@ -658,6 +665,13 @@ func (c *Conn) handleIncomingPacket(buf []byte) (*alert, error) {
 			return &alert{alertLevelFatal, alertUnexpectedMessage}, fmt.Errorf("ApplicationData with epoch of 0")
 		}
 
+		if err := c.handshakeErr.load(); err != nil {
+			return nil, err
+		}
+		if c.connectionClosed.Err() != nil {
+			return nil, io.EOF
+		}
+
 		select {
 		case c.decrypted <- content.data:
 		case <-c.connectionClosed.Done():
@@ -772,27 +786,32 @@ func (c *Conn) getRemoteEpoch() uint16 {
 	return c.state.remoteEpoch.Load().(uint16)
 }
 
-// LocalAddr is a stub
+// LocalAddr implements net.Conn.LocalAddr
 func (c *Conn) LocalAddr() net.Addr {
 	return c.nextConn.LocalAddr()
 }
 
-// RemoteAddr is a stub
+// RemoteAddr implements net.Conn.RemoteAddr
 func (c *Conn) RemoteAddr() net.Addr {
 	return c.nextConn.RemoteAddr()
 }
 
-// SetDeadline is a stub
+// SetDeadline implements net.Conn.SetDeadline
 func (c *Conn) SetDeadline(t time.Time) error {
-	return c.nextConn.SetDeadline(t)
+	c.readDeadline.Set(t)
+	return c.SetWriteDeadline(t)
 }
 
-// SetReadDeadline is a stub
+// SetReadDeadline implements net.Conn.SetReadDeadline
 func (c *Conn) SetReadDeadline(t time.Time) error {
-	return c.nextConn.SetReadDeadline(t)
+	c.readDeadline.Set(t)
+	// Read deadline is fully managed by this layer.
+	// Don't set read deadline to underlying connection.
+	return nil
 }
 
-// SetWriteDeadline is a stub
+// SetWriteDeadline implements net.Conn.SetWriteDeadline
 func (c *Conn) SetWriteDeadline(t time.Time) error {
+	c.writeDeadline.Set(t)
 	return c.nextConn.SetWriteDeadline(t)
 }
diff --git a/go.mod b/go.mod
index 8b6552d..f140a33 100644
--- a/go.mod
+++ b/go.mod
@@ -4,6 +4,7 @@ require (
 	github.com/pion/logging v0.2.2
 	github.com/pion/transport v0.8.10
 	golang.org/x/crypto v0.0.0-20200128174031-69ecbb4d6d5d
+	golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3
 )
 
 go 1.13
diff --git a/go.sum b/go.sum
index 3349e36..1417add 100644
--- a/go.sum
+++ b/go.sum
@@ -9,6 +9,7 @@ github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81P
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20200128174031-69ecbb4d6d5d h1:9FCpayM9Egr1baVnV1SX0H87m+XB0B8S0hAMi99X/3U=
 golang.org/x/crypto v0.0.0-20200128174031-69ecbb4d6d5d/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
+golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3 h1:0GoQqolDA55aaLxZyTzK/Y2ePZzZTUrRacwib7cNsYQ=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
diff --git a/internal/dpipe/dpipe.go b/internal/dpipe/dpipe.go
deleted file mode 100644
index 721d987..0000000
--- a/internal/dpipe/dpipe.go
+++ /dev/null
@@ -1,91 +0,0 @@
-// Package dpipe provides the pipe works like datagram protocol on memory.
-package dpipe
-
-import (
-	"errors"
-	"io"
-	"net"
-	"sync"
-	"time"
-)
-
-var errNotImplemented = errors.New("not implemented yet")
-
-// Pipe creates pair of non-stream conn on memory.
-// Close of the one end doesn't make effect to the other end.
-func Pipe() (net.Conn, net.Conn) {
-	ch0 := make(chan []byte, 1000)
-	ch1 := make(chan []byte, 1000)
-	return &conn{
-			rCh:    ch0,
-			wCh:    ch1,
-			closed: make(chan struct{}),
-		}, &conn{
-			rCh:    ch1,
-			wCh:    ch0,
-			closed: make(chan struct{}),
-		}
-}
-
-type pipeAddr struct{}
-
-func (pipeAddr) Network() string { return "pipe" }
-func (pipeAddr) String() string  { return "pipe" }
-
-type conn struct {
-	rCh       chan []byte
-	wCh       chan []byte
-	closed    chan struct{}
-	closeOnce sync.Once
-}
-
-func (*conn) LocalAddr() net.Addr  { return pipeAddr{} }
-func (*conn) RemoteAddr() net.Addr { return pipeAddr{} }
-
-func (*conn) SetDeadline(t time.Time) error {
-	return errNotImplemented
-}
-func (*conn) SetReadDeadline(t time.Time) error {
-	return errNotImplemented
-}
-func (*conn) SetWriteDeadline(t time.Time) error {
-	return errNotImplemented
-}
-
-func (c *conn) Read(data []byte) (n int, err error) {
-	select {
-	case <-c.closed:
-		return 0, io.EOF
-	default:
-	}
-	select {
-	case d := <-c.rCh:
-		if len(d) <= len(data) {
-			copy(data, d)
-			return len(d), nil
-		}
-		copy(data, d[:len(data)])
-		return len(data), nil
-	case <-c.closed:
-		return 0, io.EOF
-	}
-}
-
-func (c *conn) Write(data []byte) (n int, err error) {
-	select {
-	case <-c.closed:
-		return 0, io.ErrClosedPipe
-	default:
-	}
-	select {
-	case <-c.closed:
-		return 0, io.ErrClosedPipe
-	case c.wCh <- data:
-	}
-	return len(data), nil
-}
-
-func (c *conn) Close() error {
-	c.closeOnce.Do(func() { close(c.closed) })
-	return nil
-}
diff --git a/internal/net/deadline/deadline.go b/internal/net/deadline/deadline.go
new file mode 100644
index 0000000..fb5a88e
--- /dev/null
+++ b/internal/net/deadline/deadline.go
@@ -0,0 +1,77 @@
+// Package deadline provides deadline timer used to implement
+// net.Conn compatible connection
+package deadline
+
+import (
+	"sync"
+	"time"
+)
+
+// Deadline signals updatable deadline timer.
+type Deadline struct {
+	exceeded chan struct{}
+	stop     chan struct{}
+	stopped  chan bool
+	mu       sync.RWMutex
+}
+
+// New creates new deadline timer.
+func New() *Deadline {
+	d := &Deadline{
+		exceeded: make(chan struct{}),
+		stop:     make(chan struct{}),
+		stopped:  make(chan bool, 1),
+	}
+	d.stopped <- true
+	return d
+}
+
+// Set new deadline. Zero value means no deadline.
+func (d *Deadline) Set(t time.Time) {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	close(d.stop)
+
+	select {
+	case <-d.exceeded:
+		d.exceeded = make(chan struct{})
+	default:
+		stopped := <-d.stopped
+		if !stopped {
+			d.exceeded = make(chan struct{})
+		}
+	}
+	d.stop = make(chan struct{})
+	d.stopped = make(chan bool, 1)
+
+	if t.IsZero() {
+		d.stopped <- true
+		return
+	}
+
+	if dur := time.Until(t); dur > 0 {
+		exceeded := d.exceeded
+		stopped := d.stopped
+		go func() {
+			select {
+			case <-time.After(dur):
+				close(exceeded)
+				stopped <- false
+			case <-d.stop:
+				stopped <- true
+			}
+		}()
+		return
+	}
+
+	close(d.exceeded)
+	d.stopped <- false
+}
+
+// Done receives deadline signal.
+func (d *Deadline) Done() <-chan struct{} {
+	d.mu.RLock()
+	defer d.mu.RUnlock()
+	return d.exceeded
+}
diff --git a/internal/net/dpipe/dpipe.go b/internal/net/dpipe/dpipe.go
new file mode 100644
index 0000000..475de08
--- /dev/null
+++ b/internal/net/dpipe/dpipe.go
@@ -0,0 +1,142 @@
+// Package dpipe provides the pipe works like datagram protocol on memory.
+package dpipe
+
+import (
+	"context"
+	"io"
+	"net"
+	"sync"
+	"time"
+
+	"github.com/pion/dtls/v2/internal/net/deadline"
+)
+
+// Pipe creates pair of non-stream conn on memory.
+// Close of the one end doesn't make effect to the other end.
+func Pipe() (net.Conn, net.Conn) {
+	ch0 := make(chan []byte, 1000)
+	ch1 := make(chan []byte, 1000)
+	return &conn{
+			rCh:           ch0,
+			wCh:           ch1,
+			closed:        make(chan struct{}),
+			closing:       make(chan struct{}),
+			readDeadline:  deadline.New(),
+			writeDeadline: deadline.New(),
+		}, &conn{
+			rCh:           ch1,
+			wCh:           ch0,
+			closed:        make(chan struct{}),
+			closing:       make(chan struct{}),
+			readDeadline:  deadline.New(),
+			writeDeadline: deadline.New(),
+		}
+}
+
+type pipeAddr struct{}
+
+func (pipeAddr) Network() string { return "pipe" }
+func (pipeAddr) String() string  { return "pipe" }
+
+type conn struct {
+	rCh       chan []byte
+	wCh       chan []byte
+	closed    chan struct{}
+	closing   chan struct{}
+	closeOnce sync.Once
+
+	readDeadline  *deadline.Deadline
+	writeDeadline *deadline.Deadline
+}
+
+func (*conn) LocalAddr() net.Addr  { return pipeAddr{} }
+func (*conn) RemoteAddr() net.Addr { return pipeAddr{} }
+
+func (c *conn) SetDeadline(t time.Time) error {
+	c.readDeadline.Set(t)
+	c.writeDeadline.Set(t)
+	return nil
+}
+func (c *conn) SetReadDeadline(t time.Time) error {
+	c.readDeadline.Set(t)
+	return nil
+}
+func (c *conn) SetWriteDeadline(t time.Time) error {
+	c.writeDeadline.Set(t)
+	return nil
+}
+
+func (c *conn) Read(data []byte) (n int, err error) {
+	select {
+	case <-c.closed:
+		return 0, io.EOF
+	case <-c.closing:
+		if len(c.rCh) == 0 {
+			return 0, io.EOF
+		}
+	case <-c.readDeadline.Done():
+		return 0, context.DeadlineExceeded
+	default:
+	}
+
+	for {
+		select {
+		case d := <-c.rCh:
+			if len(d) <= len(data) {
+				copy(data, d)
+				return len(d), nil
+			}
+			copy(data, d[:len(data)])
+			return len(data), nil
+		case <-c.closed:
+			return 0, io.EOF
+		case <-c.closing:
+			if len(c.rCh) == 0 {
+				return 0, io.EOF
+			}
+		case <-c.readDeadline.Done():
+			return 0, context.DeadlineExceeded
+		}
+	}
+}
+
+func (c *conn) cleanWriteBuffer() {
+	for {
+		select {
+		case <-c.wCh:
+		default:
+			return
+		}
+	}
+}
+
+func (c *conn) Write(data []byte) (n int, err error) {
+	select {
+	case <-c.closed:
+		return 0, io.ErrClosedPipe
+	case <-c.writeDeadline.Done():
+		c.cleanWriteBuffer()
+		return 0, context.DeadlineExceeded
+	default:
+	}
+
+	cData := make([]byte, len(data))
+	copy(cData, data)
+
+	select {
+	case <-c.closed:
+		return 0, io.ErrClosedPipe
+	case <-c.writeDeadline.Done():
+		c.cleanWriteBuffer()
+		return 0, context.DeadlineExceeded
+	case c.wCh <- cData:
+		return len(cData), nil
+	}
+}
+
+func (c *conn) Close() error {
+	c.closeOnce.Do(func() {
+		close(c.closed)
+	})
+	return nil
+}
diff --git a/internal/udp/conn.go b/internal/net/udp/conn.go
similarity index 82%
rename from internal/udp/conn.go
rename to internal/net/udp/conn.go
index ab65300..b67c50f 100644
--- a/internal/udp/conn.go
+++ b/internal/net/udp/conn.go
@@ -2,12 +2,15 @@
 package udp
 
 import (
+	"context"
 	"errors"
 	"io"
 	"net"
 	"sync"
 	"sync/atomic"
 	"time"
+
+	"github.com/pion/dtls/v2/internal/net/deadline"
 )
 
 const receiveMTU = 8192
@@ -153,15 +156,20 @@ type Conn struct {
 
 	doneCh   chan struct{}
 	doneOnce sync.Once
+
+	readDeadline  *deadline.Deadline
+	writeDeadline *deadline.Deadline
 }
 
 func (l *Listener) newConn(rAddr net.Addr) *Conn {
 	return &Conn{
-		listener: l,
-		rAddr:    rAddr,
-		readCh:   make(chan []byte),
-		sizeCh:   make(chan int),
-		doneCh:   make(chan struct{}),
+		listener:      l,
+		rAddr:         rAddr,
+		readCh:        make(chan []byte),
+		sizeCh:        make(chan int),
+		doneCh:        make(chan struct{}),
+		readDeadline:  deadline.New(),
+		writeDeadline: deadline.New(),
 	}
 }
 
@@ -173,11 +181,18 @@ func (c *Conn) Read(p []byte) (int, error) {
 		return n, nil
 	case <-c.doneCh:
 		return 0, io.EOF
+	case <-c.readDeadline.Done():
+		return 0, context.DeadlineExceeded
 	}
 }
 
 // Write writes len(p) bytes from p to the DTLS connection
 func (c *Conn) Write(p []byte) (n int, err error) {
+	select {
+	case <-c.writeDeadline.Done():
+		return 0, context.DeadlineExceeded
+	default:
+	}
 	return c.listener.pConn.WriteTo(p, c.rAddr)
 }
 
@@ -206,27 +221,33 @@ func (c *Conn) Close() error {
 	return err
 }
 
-// LocalAddr is a stub
+// LocalAddr implements net.Conn.LocalAddr
 func (c *Conn) LocalAddr() net.Addr {
 	return c.listener.pConn.LocalAddr()
 }
 
-// RemoteAddr is a stub
+// RemoteAddr implements net.Conn.RemoteAddr
 func (c *Conn) RemoteAddr() net.Addr {
 	return c.rAddr
 }
 
-// SetDeadline is a stub
+// SetDeadline implements net.Conn.SetDeadline
 func (c *Conn) SetDeadline(t time.Time) error {
+	c.readDeadline.Set(t)
+	c.writeDeadline.Set(t)
+	// Deadline of underlying connection should not be changed
+	// since the connection can be shared.
 	return nil
 }
 
-// SetReadDeadline is a stub
+// SetReadDeadline implements net.Conn.SetDeadline
 func (c *Conn) SetReadDeadline(t time.Time) error {
+	c.readDeadline.Set(t)
 	return nil
 }
 
-// SetWriteDeadline is a stub
+// SetWriteDeadline implements net.Conn.SetDeadline
 func (c *Conn) SetWriteDeadline(t time.Time) error {
+	c.writeDeadline.Set(t)
 	return nil
 }
diff --git a/listener.go b/listener.go
index d5151c6..732cf76 100644
--- a/listener.go
+++ b/listener.go
@@ -3,7 +3,7 @@ package dtls
 import (
 	"net"
 
-	"github.com/pion/dtls/v2/internal/udp"
+	"github.com/pion/dtls/v2/internal/net/udp"
 )
 
 // Listen creates a DTLS listener


Test patch:
diff --git a/conn_test.go b/conn_test.go
index 228c2b3..66397e6 100644
--- a/conn_test.go
+++ b/conn_test.go
@@ -11,29 +11,11 @@ import (
 	"testing"
 	"time"
 
-	"github.com/pion/dtls/v2/internal/dpipe"
+	"github.com/pion/dtls/v2/internal/net/dpipe"
 	"github.com/pion/dtls/v2/pkg/crypto/selfsign"
 	"github.com/pion/transport/test"
 )
 
-// Seems to strict for out implementation at this point
-// func TestNetTest(t *testing.T) {
-// 	lim := test.TimeOut(time.Minute*1 + time.Second*10)
-// 	defer lim.Stop()
-//
-// 	nettest.TestConn(t, func() (c1, c2 net.Conn, stop func(), err error) {
-// 		c1, c2, err = pipeMemory()
-// 		if err != nil {
-// 			return nil, nil, nil, err
-// 		}
-// 		stop = func() {
-// 			c1.Close()
-// 			c2.Close()
-// 		}
-// 		return
-// 	})
-// }
-
 func TestStressDuplex(t *testing.T) {
 	// Limit runtime in case of deadlocks
 	lim := test.TimeOut(time.Second * 20)
diff --git a/internal/net/deadline/deadline_test.go b/internal/net/deadline/deadline_test.go
new file mode 100644
index 0000000..7df8774
--- /dev/null
+++ b/internal/net/deadline/deadline_test.go
@@ -0,0 +1,123 @@
+package deadline
+
+import (
+	"bytes"
+	"context"
+	"testing"
+	"time"
+)
+
+func TestDeadline(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	t.Run("Deadline", func(t *testing.T) {
+		now := time.Now()
+
+		ctx0, cancel0 := context.WithDeadline(ctx, now.Add(40*time.Millisecond))
+		defer cancel0()
+		ctx1, cancel1 := context.WithDeadline(ctx, now.Add(60*time.Millisecond))
+		defer cancel1()
+		d := New()
+		d.Set(now.Add(50 * time.Millisecond))
+
+		ch := make(chan byte)
+		go sendOnDone(ctx, ctx0.Done(), ch, 0)
+		go sendOnDone(ctx, ctx1.Done(), ch, 1)
+		go sendOnDone(ctx, d.Done(), ch, 2)
+
+		calls := collectCh(ch, 3, 100*time.Millisecond)
+		expectedCalls := []byte{0, 2, 1}
+		if !bytes.Equal(calls, expectedCalls) {
+			t.Errorf("Wrong order of deadline signal, expected: %v, got: %v", expectedCalls, calls)
+		}
+	})
+
+	t.Run("DeadlineExtend", func(t *testing.T) {
+		now := time.Now()
+
+		ctx0, cancel0 := context.WithDeadline(ctx, now.Add(40*time.Millisecond))
+		defer cancel0()
+		ctx1, cancel1 := context.WithDeadline(ctx, now.Add(60*time.Millisecond))
+		defer cancel1()
+		d := New()
+		d.Set(now.Add(50 * time.Millisecond))
+		d.Set(now.Add(70 * time.Millisecond))
+
+		ch := make(chan byte)
+		go sendOnDone(ctx, ctx0.Done(), ch, 0)
+		go sendOnDone(ctx, ctx1.Done(), ch, 1)
+		go sendOnDone(ctx, d.Done(), ch, 2)
+
+		calls := collectCh(ch, 3, 100*time.Millisecond)
+		expectedCalls := []byte{0, 1, 2}
+		if !bytes.Equal(calls, expectedCalls) {
+			t.Errorf("Wrong order of deadline signal, expected: %v, got: %v", expectedCalls, calls)
+		}
+	})
+
+	t.Run("DeadlinePretend", func(t *testing.T) {
+		now := time.Now()
+
+		ctx0, cancel0 := context.WithDeadline(ctx, now.Add(40*time.Millisecond))
+		defer cancel0()
+		ctx1, cancel1 := context.WithDeadline(ctx, now.Add(60*time.Millisecond))
+		defer cancel1()
+		d := New()
+		d.Set(now.Add(50 * time.Millisecond))
+		d.Set(now.Add(30 * time.Millisecond))
+
+		ch := make(chan byte)
+		go sendOnDone(ctx, ctx0.Done(), ch, 0)
+		go sendOnDone(ctx, ctx1.Done(), ch, 1)
+		go sendOnDone(ctx, d.Done(), ch, 2)
+
+		calls := collectCh(ch, 3, 100*time.Millisecond)
+		expectedCalls := []byte{2, 0, 1}
+		if !bytes.Equal(calls, expectedCalls) {
+			t.Errorf("Wrong order of deadline signal, expected: %v, got: %v", expectedCalls, calls)
+		}
+	})
+
+	t.Run("DeadlineCancel", func(t *testing.T) {
+		now := time.Now()
+
+		ctx0, cancel0 := context.WithDeadline(ctx, now.Add(40*time.Millisecond))
+		defer cancel0()
+		d := New()
+		d.Set(now.Add(50 * time.Millisecond))
+		d.Set(time.Time{})
+
+		ch := make(chan byte)
+		go sendOnDone(ctx, ctx0.Done(), ch, 0)
+		go sendOnDone(ctx, d.Done(), ch, 1)
+
+		calls := collectCh(ch, 2, 60*time.Millisecond)
+		expectedCalls := []byte{0}
+		if !bytes.Equal(calls, expectedCalls) {
+			t.Errorf("Wrong order of deadline signal, expected: %v, got: %v", expectedCalls, calls)
+		}
+	})
+}
+
+func sendOnDone(ctx context.Context, done <-chan struct{}, dest chan byte, val byte) {
+	select {
+	case <-done:
+	case <-ctx.Done():
+		return
+	}
+	dest <- val
+}
+func collectCh(ch <-chan byte, n int, timeout time.Duration) []byte {
+	a := time.After(timeout)
+	var calls []byte
+	for len(calls) < n {
+		select {
+		case call := <-ch:
+			calls = append(calls, call)
+		case <-a:
+			return calls
+		}
+	}
+	return calls
+}
diff --git a/internal/dpipe/dpipe_test.go b/internal/net/dpipe/dpipe_test.go
similarity index 81%
rename from internal/dpipe/dpipe_test.go
rename to internal/net/dpipe/dpipe_test.go
index 6cba457..c3fed56 100644
--- a/internal/dpipe/dpipe_test.go
+++ b/internal/net/dpipe/dpipe_test.go
@@ -8,8 +8,32 @@ import (
 	"net"
 	"testing"
 	"time"
+
+	"golang.org/x/net/nettest"
 )
 
+func TestNetTest(t *testing.T) {
+	nettest.TestConn(t, func() (net.Conn, net.Conn, func(), error) {
+		ca, cb := Pipe()
+		return &closePropagator{ca.(*conn), cb.(*conn)},
+			&closePropagator{cb.(*conn), ca.(*conn)},
+			func() {
+				_ = ca.Close()
+				_ = cb.Close()
+			}, nil
+	})
+}
+
+type closePropagator struct {
+	*conn
+	otherEnd *conn
+}
+
+func (c *closePropagator) Close() error {
+	close(c.otherEnd.closing)
+	return c.conn.Close()
+}
+
 func TestPipe(t *testing.T) {
 	ca, cb := Pipe()
 
diff --git a/internal/udp/conn_test.go b/internal/net/udp/conn_test.go
similarity index 98%
rename from internal/udp/conn_test.go
rename to internal/net/udp/conn_test.go
index 91178d7..615de27 100644
--- a/internal/udp/conn_test.go
+++ b/internal/net/udp/conn_test.go
@@ -101,7 +101,7 @@ func TestListenerCloseTimeout(t *testing.T) {
 	}
 }
 
-func pipe() (*Listener, *Conn, *net.UDPConn, error) {
+func pipe() (*Listener, net.Conn, *net.UDPConn, error) {
 	// Start listening
 	network, addr := getConfig()
 	listener, err := Listen(network, addr)
@@ -124,7 +124,7 @@ func pipe() (*Listener, *Conn, *net.UDPConn, error) {
 	}
 
 	// Accept the connection
-	var lConn *Conn
+	var lConn net.Conn
 	lConn, err = listener.Accept()
 	if err != nil {
 		return nil, nil, nil, fmt.Errorf("failed to accept Conn: %v", err)
diff --git a/nettest_test.go b/nettest_test.go
new file mode 100644
index 0000000..a0a0717
--- /dev/null
+++ b/nettest_test.go
@@ -0,0 +1,30 @@
+// +build !js
+
+package dtls
+
+import (
+	"net"
+	"testing"
+	"time"
+
+	"golang.org/x/net/nettest"
+
+	"github.com/pion/transport/test"
+)
+
+func TestNetTest(t *testing.T) {
+	lim := test.TimeOut(time.Minute*1 + time.Second*10)
+	defer lim.Stop()
+
+	nettest.TestConn(t, func() (c1, c2 net.Conn, stop func(), err error) {
+		c1, c2, err = pipeMemory()
+		if err != nil {
+			return nil, nil, nil, err
+		}
+		stop = func() {
+			_ = c1.Close()
+			_ = c2.Close()
+		}
+		return
+	})
+}