Repository: pion/dtls
Pull request #164 (2019-12-18 12:15:49)

Add datagram-like pipe for connection tests
Add on-memory pipe that closing the one end doesn't make effect to
the other end. This simulates UDP like connection.

related to https://github.com/pion/dtls/issues/163#issuecomment-566950455

codecov-io: # [Codecov](https://codecov.io/gh/pion/dtls/pull/164?src=pr&el=h1) Report
> Merging [#164](https://codecov.io/gh/pion/dtls/pull/164?src=pr&el=desc) into [master](https://codecov.io/gh/pion/dtls/commit/486ad1d14fc4932346dd3620f075b2cc32d64b4b?src=pr&el=desc) will **decrease** coverage by `<.01%`.
> The diff coverage is `75%`.

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

```diff
@@            Coverage Diff             @@
##           master     #164      +/-   ##
==========================================
- Coverage   73.38%   73.37%   -0.01%     
==========================================
  Files          58       59       +1     
  Lines        3340     3388      +48     
==========================================
+ Hits         2451     2486      +35     
- Misses        620      634      +14     
+ Partials      269      268       -1
```


| [Impacted Files](https://codecov.io/gh/pion/dtls/pull/164?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [internal/dpipe/dpipe.go](https://codecov.io/gh/pion/dtls/pull/164/diff?src=pr&el=tree#diff-aW50ZXJuYWwvZHBpcGUvZHBpcGUuZ28=) | `75% <75%> (ø)` | |
| [client\_handlers.go](https://codecov.io/gh/pion/dtls/pull/164/diff?src=pr&el=tree#diff-Y2xpZW50X2hhbmRsZXJzLmdv) | `82.59% <0%> (-0.56%)` | :arrow_down: |
| [conn.go](https://codecov.io/gh/pion/dtls/pull/164/diff?src=pr&el=tree#diff-Y29ubi5nbw==) | `80.04% <0%> (+0.23%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/pion/dtls/pull/164?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/164?src=pr&el=footer). Last update [486ad1d...7e5b342](https://codecov.io/gh/pion/dtls/pull/164?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

at-wat: It once failed in `TestStressDuplex` at https://github.com/pion/dtls/blob/7e5b342c1d2abf710db1e8725586ea7b695ebd41/conn_test.go#L58
This is what https://github.com/pion/dtls/issues/163 going to fix.
daenney: Yeah, I noticed and retriggered the build. Passed this time.
Sean-Der: amazing @at-wat! Thank you so much for grabbing, this will make our error work so much easier :)

Patch:
diff --git a/internal/dpipe/dpipe.go b/internal/dpipe/dpipe.go
new file mode 100644
index 0000000..721d987
--- /dev/null
+++ b/internal/dpipe/dpipe.go
@@ -0,0 +1,91 @@
+// 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
+}


Test patch:
diff --git a/conn_test.go b/conn_test.go
index a904e8d..7b1b0aa 100644
--- a/conn_test.go
+++ b/conn_test.go
@@ -11,6 +11,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/pion/dtls/v2/internal/dpipe"
 	"github.com/pion/transport/test"
 )
 
@@ -102,7 +103,7 @@ func TestRoutineLeakOnClose(t *testing.T) {
 
 func pipeMemory() (*Conn, *Conn, error) {
 	// In memory pipe
-	ca, cb := net.Pipe()
+	ca, cb := dpipe.Pipe()
 
 	type result struct {
 		c   *Conn
@@ -163,7 +164,7 @@ func TestHandshakeWithAlert(t *testing.T) {
 
 	clientErr := make(chan error, 1)
 
-	ca, cb := net.Pipe()
+	ca, cb := dpipe.Pipe()
 	go func() {
 		conf := &Config{
 			CipherSuites: []CipherSuiteID{TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256},
@@ -258,7 +259,7 @@ func TestPSK(t *testing.T) {
 		clientIdentity := []byte("Client Identity")
 		clientErr := make(chan error, 1)
 
-		ca, cb := net.Pipe()
+		ca, cb := dpipe.Pipe()
 		go func() {
 			conf := &Config{
 				PSK: func(hint []byte) ([]byte, error) {
@@ -307,7 +308,7 @@ func TestPSKHintFail(t *testing.T) {
 
 	clientErr := make(chan error, 1)
 
-	ca, cb := net.Pipe()
+	ca, cb := dpipe.Pipe()
 	go func() {
 		conf := &Config{
 			PSK: func(hint []byte) ([]byte, error) {
@@ -345,7 +346,7 @@ func TestClientTimeout(t *testing.T) {
 
 	clientErr := make(chan error, 1)
 
-	ca, _ := net.Pipe()
+	ca, _ := dpipe.Pipe()
 	go func() {
 		conf := &Config{
 			ConnectTimeout: ConnectTimeoutOption(1 * time.Second),
@@ -404,7 +405,7 @@ func TestSRTPConfiguration(t *testing.T) {
 			WantServerError: nil,
 		},
 	} {
-		ca, cb := net.Pipe()
+		ca, cb := dpipe.Pipe()
 		type result struct {
 			c   *Conn
 			err error
@@ -552,7 +553,7 @@ func TestClientCertificate(t *testing.T) {
 	for name, tt := range tests {
 		tt := tt
 		t.Run(name, func(t *testing.T) {
-			ca, cb := net.Pipe()
+			ca, cb := dpipe.Pipe()
 			type result struct {
 				c   *Conn
 				err error
@@ -711,7 +712,7 @@ func TestExtendedMasterSecret(t *testing.T) {
 	for name, tt := range tests {
 		tt := tt
 		t.Run(name, func(t *testing.T) {
-			ca, cb := net.Pipe()
+			ca, cb := dpipe.Pipe()
 			type result struct {
 				c   *Conn
 				err error
@@ -815,7 +816,7 @@ func TestServerCertificate(t *testing.T) {
 	for name, tt := range tests {
 		tt := tt
 		t.Run(name, func(t *testing.T) {
-			ca, cb := net.Pipe()
+			ca, cb := dpipe.Pipe()
 			go func() {
 				_, _ = Server(cb, tt.serverCfg)
 			}()
@@ -882,7 +883,7 @@ func TestCipherSuiteConfiguration(t *testing.T) {
 			WantServerError:    nil,
 		},
 	} {
-		ca, cb := net.Pipe()
+		ca, cb := dpipe.Pipe()
 		type result struct {
 			c   *Conn
 			err error
@@ -967,7 +968,7 @@ func TestPSKConfiguration(t *testing.T) {
 			WantServerError:      errServerMustHaveCertificate,
 		},
 	} {
-		ca, cb := net.Pipe()
+		ca, cb := dpipe.Pipe()
 		type result struct {
 			c   *Conn
 			err error
@@ -1058,7 +1059,7 @@ func TestServerTimeout(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	ca, cb := net.Pipe()
+	ca, cb := dpipe.Pipe()
 	defer func() {
 		err := ca.Close()
 		if err != nil {
diff --git a/internal/dpipe/dpipe_test.go b/internal/dpipe/dpipe_test.go
new file mode 100644
index 0000000..0ae43c9
--- /dev/null
+++ b/internal/dpipe/dpipe_test.go
@@ -0,0 +1,79 @@
+package dpipe
+
+import (
+	"bytes"
+	"io"
+	"net"
+	"testing"
+	"time"
+)
+
+func TestPipe(t *testing.T) {
+	ca, cb := Pipe()
+
+	testData := []byte{0x01, 0x02}
+
+	for name, cond := range map[string]struct {
+		ca net.Conn
+		cb net.Conn
+	}{
+		"AtoB": {ca, cb},
+		"BtoA": {cb, ca},
+	} {
+		c0 := cond.ca
+		c1 := cond.cb
+		t.Run(name, func(t *testing.T) {
+			switch n, err := c0.Write(testData); {
+			case err != nil:
+				t.Errorf("Unexpected error on Write: %v", err)
+			case n != len(testData):
+				t.Errorf("Expected to write %d bytes, wrote %d bytes", len(testData), n)
+			}
+
+			readData := make([]byte, 4)
+			switch n, err := c1.Read(readData); {
+			case err != nil:
+				t.Errorf("Unexpected error on Write: %v", err)
+			case n != len(testData):
+				t.Errorf("Expected to read %d bytes, got %d bytes", len(testData), n)
+			case !bytes.Equal(testData, readData[0:n]):
+				t.Errorf("Expected to read %v, got %v", testData, readData[0:n])
+			}
+		})
+	}
+
+	if err := ca.Close(); err != nil {
+		t.Errorf("Unexpected error on Close: %v", err)
+	}
+	if _, err := ca.Write(testData); err != io.ErrClosedPipe {
+		t.Errorf("Write to closed conn should fail with %v, got %v", io.ErrClosedPipe, err)
+	}
+
+	// Other side should be writable.
+	if _, err := cb.Write(testData); err != nil {
+		t.Errorf("Unexpected error on Write: %v", err)
+	}
+
+	readData := make([]byte, 4)
+	if _, err := ca.Read(readData); err != io.EOF {
+		t.Errorf("Read from closed conn should fail with %v, got %v", io.EOF, err)
+	}
+
+	// Other side should be readable.
+	readDone := make(chan struct{})
+	go func() {
+		readData := make([]byte, 4)
+		if n, err := cb.Read(readData); err == nil {
+			t.Errorf("Unexpected data %v was arrived to orphaned conn", readData[:n])
+		}
+		close(readDone)
+	}()
+	select {
+	case <-readDone:
+		t.Errorf("Read should be blocked if the other side is closed")
+	case <-time.After(10 * time.Millisecond):
+	}
+	if err := cb.Close(); err != nil {
+		t.Errorf("Unexpected error on Close: %v", err)
+	}
+}