runtime/internal/rpc/stream/manager/xmanager_test.go

// Copyright 2015 The Vanadium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

package manager_test

import (
	"bytes"
	"fmt"
	"io"
	"os"
	"strings"
	"testing"

	"v.io/v23/context"
	"v.io/v23/naming"
	"v.io/v23/vtrace"
	"v.io/x/ref/lib/flags"
	"v.io/x/ref/runtime/internal/rpc/proxy"
	"v.io/x/ref/runtime/internal/rpc/stream"
	"v.io/x/ref/runtime/internal/rpc/stream/manager"
	ivtrace "v.io/x/ref/runtime/internal/vtrace"
	"v.io/x/ref/test/testutil"
)

func Init() (*context.T, context.CancelFunc) {
	ctx, cancel := context.RootContext()
	ctx, err := ivtrace.Init(ctx, flags.VtraceFlags{
		CollectRegexp:  ".*",
		DumpOnShutdown: true,
		CacheSize:      100,
	})
	if err != nil {
		panic(err)
	}
	return ctx, func() {
		vtrace.FormatTraces(os.Stderr, vtrace.GetStore(ctx).TraceRecords(), nil)
		cancel()
	}
}

func TestDirectConnection(t *testing.T) {
	ctx, cancel := Init()
	defer cancel()

	p := testutil.NewPrincipal("test")
	rid := naming.FixedRoutingID(0x5555)
	m := manager.XInternalNew(ctx, rid)
	want := "read this please"

	ln, ep, err := m.Listen("tcp", "127.0.0.1:0", p, p.BlessingStore().Default())
	if err != nil {
		t.Error(err)
	}

	go func(ln stream.XListener) {
		flow, err := ln.Accept()
		if err != nil {
			t.Error(err)
		}
		got, err := readLine(flow)
		if err != nil {
			t.Error(err)
		}
		if got != want {
			t.Errorf("got %v, want %v", got, want)
		}
		ln.Close()
	}(ln)

	flow, err := m.Dial(ep, p)
	if err != nil {
		t.Error(err)
	}
	writeLine(flow, want)
}

func TestProxyConnection(t *testing.T) {
	ctx, cancel := Init()
	defer cancel()

	p := testutil.NewPrincipal("test")
	srid := naming.FixedRoutingID(0x5555)
	crid := naming.FixedRoutingID(0x4444)
	prid := naming.FixedRoutingID(0x6666)
	sm := manager.XInternalNew(ctx, srid)
	cm := manager.XInternalNew(ctx, crid)
	want := "read this please"

	_, sd, pep, err := proxy.New("tcp", "127.0.0.1:0", prid, ctx, p)
	if err != nil {
		t.Fatal(err)
	}
	defer sd()
	fmt.Printf("proxy ep: %v\n", pep)

	// listen on the proxy
	ln, ep, err := sm.Listen("v23", pep.String(), p, p.BlessingStore().Default())
	fmt.Printf("server ep: %v\n", ep)
	if err != nil {
		t.Error(err)
	}

	go func(ln stream.XListener) {
		flow, err := ln.Accept()
		if err != nil {
			t.Error(err)
		}
		got, err := readLine(flow)
		if err != nil {
			t.Error(err)
		}
		if got != want {
			t.Errorf("got %v, want %v", got, want)
		}
		ln.Close()
	}(ln)

	flow, err := cm.Dial(ep, p)
	if err != nil {
		t.Error(err)
	}

	writeLine(flow, want)
}

func readLine(f stream.XFlow) (string, error) {
	var result bytes.Buffer
	var buf [5]byte
	for {
		n, err := f.Read(buf[:])
		result.Write(buf[:n])
		if err == io.EOF || buf[n-1] == '\n' {
			return strings.TrimRight(result.String(), "\n"), nil
		}
		if err != nil {
			return "", fmt.Errorf("Read returned (%d, %v)", n, err)
		}
	}
}

func writeLine(f stream.XFlow, data string) error {
	data = data + "\n"
	if n, err := f.Write([]byte(data)); err != nil {
		return fmt.Errorf("Write returned (%d, %v)", n, err)
	}
	return nil
}