Compare minq and golang TCP IO discipline

minq_test.go

package minhq_test

import (
	"io"
	"net"
	"testing"

	"github.com/ekr/minq"
	"github.com/stvp/assert"
)

type transport struct {
	read  <-chan []byte
	write chan<- []byte
}

func (t *transport) Send(p []byte) error {
	t.write <- p
	return nil
}

func newTransportPair() (*transport, *transport) {
	a := make(chan []byte, 10)
	b := make(chan []byte, 10)
	return &transport{a, b}, &transport{b, a}
}

type transportFactory struct {
	t *transport
}

func (tf *transportFactory) MakeTransport(remote *net.UDPAddr) (minq.Transport, error) {
	transport := tf.t
	tf.t = nil
	return transport, nil
}

type connectionHandler struct {
	streams chan *minq.Stream
}

func makeConnectionHandler() *connectionHandler {
	return &connectionHandler{make(chan *minq.Stream)}
}
func (ch *connectionHandler) StateChanged(s minq.State) {}
func (ch *connectionHandler) NewStream(s *minq.Stream) {
	ch.streams <- s
}
func (ch *connectionHandler) StreamReadable(s *minq.Stream) {}

type peer struct {
	// The connection.
	c *minq.Connection
	// The receive transport.
	t *transport
	// This receives events for the connection.
	ch *connectionHandler
}

func (p *peer) setConnection(c *minq.Connection) {
	p.c = c
	p.ch = makeConnectionHandler()
	p.c.SetHandler(p.ch)
}

func (p *peer) service(s func(minq.State) bool) {
	if s == nil {
		s = func(s minq.State) bool {
			return s != minq.StateEstablished
		}
	}
	for !s(p.c.GetState()) {
		var err error
		select {
		case b := <-p.t.read:
			err = p.c.Input(b)
			break

		default:
			_, err = p.c.CheckTimer()
			break
		}
		if err != nil {
			return
		}
	}
}

func (p *peer) handshake() {
	p.service(func(s minq.State) bool {
		return s == minq.StateEstablished ||
			s == minq.StateError
	})
}

func setup() (*peer, *peer) {
	testTlsConfig := minq.NewTlsConfig("localhost")
	clientAddr := &net.UDPAddr{IP: net.ParseIP("::1"), Port: 12589}
	var client peer
	var server peer
	client.t, server.t = newTransportPair()

	connection := make(chan *peer)
	go func() {
		c := minq.NewConnection(client.t, minq.RoleClient, &testTlsConfig, nil)
		client.setConnection(c)
		client.handshake()
		connection <- &client
		client.service(nil)
	}()

	go func() {
		s := minq.NewServer(&transportFactory{server.t}, &testTlsConfig, nil)
		for server.c == nil {
			c, err := s.Input(clientAddr, <-server.t.read)
			if err != nil {
				connection <- nil
				return
			}
			server.setConnection(c)
			c.SetHandler(server.ch)
		}
		server.handshake()
		connection <- &server
		server.service(nil)
	}()

	return <-connection, <-connection
}

func TestConnect(t *testing.T) {
	client, server := setup()

	cstr := client.c.CreateStream()
	out := []byte{1, 2, 3}
	n, err := cstr.Write(out)
	assert.Nil(t, err)
	assert.Equal(t, 3, n)

	sstr := <-server.ch.streams
	assert.Equal(t, cstr.Id(), sstr.Id())

	in := make([]byte, len(out))
	n, err = sstr.Read(in)
	assert.Nil(t, err)
	assert.Equal(t, 3, n)
	assert.Equal(t, out, in)

  // This returns minq.ErrorWouldBlock rather than blocking.
	/*n, err = sstr.Read(in)
	assert.Nil(t, err) */
}

func TestTCP(t *testing.T) {
	done := make(chan *net.TCPConn)
	server, err := net.ListenTCP("tcp", nil)
	assert.Nil(t, err)
	go func() {
		scon, err := server.AcceptTCP()
		assert.Nil(t, err)
		done <- scon
	}()
	go func() {
		client, err := net.DialTCP("tcp", nil, server.Addr().(*net.TCPAddr))
		assert.Nil(t, err)
		done <- client
	}()
	a := <-done
	b := <-done

	out := []byte{1, 2, 3}
	n, err := a.Write(out)
	assert.Nil(t, err)
	assert.Equal(t, len(out), n)

	in := make([]byte, len(out))
	n, err = b.Read(in)
	assert.Nil(t, err)
	assert.Equal(t, len(out), n)
	assert.Equal(t, out, in)

	// This is weird. Socket a is garbage collected at this point if we don't refer
	// to it any more. At that point, it sends a FIN and this returns EOF. But if
	// we uncomment the block below, the read will block and therefore hang.
	n, err = b.Read(in)
	assert.Equal(t, io.EOF, err)

	/*n, err = a.Write(out)
	assert.Nil(t, err)
	assert.Equal(t, len(out), n)

	n, err = b.Read(in)
	assert.Nil(t, err)
	assert.Equal(t, len(out), n)
	assert.Equal(t, out, in)*/
}