raw_stream_test.go RawHistoryFolder

1package brokeripc
2
3import (
4	"context"
5	"encoding/binary"
6	"io"
7	"testing"
8	"time"
9
10	"github.com/stretchr/testify/require"
11)
12
13type rawRelayCore struct {
14	received chan []byte
15	done     chan struct{}
16}
17
18func (c *rawRelayCore) SendEnvelope(payload []byte) error {
19	c.received <- append([]byte(nil), payload...)
20	return nil
21}
22func (c *rawRelayCore) RecvEnvelope() ([]byte, error) { <-c.done; return nil, io.EOF }
23func (c *rawRelayCore) Done() <-chan struct{}         { return c.done }
24func (c *rawRelayCore) Err() error                    { return nil }
25func (c *rawRelayCore) Close() error {
26	select {
27	case <-c.done:
28	default:
29		close(c.done)
30	}
31	return nil
32}
33
34func TestServerStreamRelaysOpaqueEnvelopes(t *testing.T) {
35	core := &rawRelayCore{received: make(chan []byte, 1), done: make(chan struct{})}
36	session := &serverSession{ctx: context.Background(), cfg: Config{}.withDefaults()}
37	stream, err := newServerStream(session, 1, core)
38	require.NoError(t, err)
39	session.wg.Add(1)
40	go stream.relayClient()
41	payload := []byte{0xff, 0xfe, 0xfd, 0x80}
42	frame := make([]byte, 4+len(payload))
43	binary.BigEndian.PutUint32(frame[:4], uint32(len(payload)))
44	copy(frame[4:], payload)
45	require.NoError(t, stream.deliver(frame[:2]))
46	require.NoError(t, stream.deliver(frame[2:]))
47	select {
48	case got := <-core.received:
49		require.Equal(t, payload, got)
50	case <-time.After(5 * time.Second):
51		t.Fatal("opaque envelope not relayed")
52	}
53	stream.close(nil)
54	session.wg.Wait()
55}