lib/pubsub/config_test.go - release.go.x.ref - Git at Google

 // 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 pubsub_test

 import (
 	"fmt"
 	"math/rand"
 	"reflect"
 	"sync"
 	"testing"
 	"time"

 	"v.io/x/ref/lib/pubsub"

 	"v.io/v23/verror"
 )

 func ExamplePublisher() {
 	in := make(chan pubsub.Setting)
 	pub := pubsub.NewPublisher()
 	pub.CreateStream("net", "network settings", in)

 	// A simple producer of IP address settings.
 	producer := func() {
 		in <- pubsub.NewString("ip", "address", "1.2.3.5")
 	}

 	var waiter sync.WaitGroup
 	waiter.Add(2)

 	// A simple consumer of IP address Settings.
 	consumer := func(ch chan pubsub.Setting) {
 		fmt.Println(<-ch)
 		waiter.Done()
 	}

 	// Publish an initial Setting to the Stream.
 	in <- pubsub.NewString("ip", "address", "1.2.3.4")

 	// Fork the stream twice, and read the latest value.
 	ch1 := make(chan pubsub.Setting)
 	st, _ := pub.ForkStream("net", ch1)
 	fmt.Println(st.Latest["ip"])
 	ch2 := make(chan pubsub.Setting)
 	st, _ = pub.ForkStream("net", ch2)
 	fmt.Println(st.Latest["ip"])

 	// Now we can read new Settings as they are generated.
 	go producer()
 	go consumer(ch1)
 	go consumer(ch2)

 	waiter.Wait()

 	// Output:
 	// ip: address: (string: 1.2.3.4)
 	// ip: address: (string: 1.2.3.4)
 	// ip: address: (string: 1.2.3.5)
 	// ip: address: (string: 1.2.3.5)
 }

 func ExampleShutdown() {
 	in := make(chan pubsub.Setting)
 	pub := pubsub.NewPublisher()
 	stop, _ := pub.CreateStream("net", "network settings", in)

 	var producerReady sync.WaitGroup
 	producerReady.Add(1)
 	var consumersReady sync.WaitGroup
 	consumersReady.Add(2)

 	// A producer to write 100 Settings before signalling that it's
 	// ready to be shutdown. This is purely to demonstrate how to use
 	// Shutdown.
 	producer := func() {
 		for i := 0; ; i++ {
 			select {
 			case <-stop:
 				close(in)
 				return
 			default:
 				in <- pubsub.NewString("ip", "address", "1.2.3.4")
 				if i == 100 {
 					producerReady.Done()
 				}
 			}
 		}
 	}

 	var waiter sync.WaitGroup
 	waiter.Add(2)

 	consumer := func() {
 		ch := make(chan pubsub.Setting, 10)
 		pub.ForkStream("net", ch)
 		consumersReady.Done()
 		i := 0
 		for {
 			if _, ok := <-ch; !ok {
 				// The channel has been closed when the publisher
 				// is asked to shut down.
 				break
 			}
 			i++
 		}
 		if i >= 100 {
 			// We've received at least 100 Settings as per the producer above.
 			fmt.Println("done")
 		}
 		waiter.Done()
 	}

 	go consumer()
 	go consumer()
 	consumersReady.Wait()
 	go producer()
 	producerReady.Wait()
 	pub.Shutdown()
 	waiter.Wait()
 	// Output:
 	// done
 	// done
 }

 func TestSimple(t *testing.T) {
 	ch := make(chan pubsub.Setting, 2)
 	pub := pubsub.NewPublisher()
 	if _, err := pub.ForkStream("stream", nil); err == nil || verror.ErrorID(err) != "v.io/x/ref/lib/pubsub.errStreamDoesntExist" {
 		t.Errorf("missing or wrong error: %v", err)
 	}
 	if _, err := pub.CreateStream("stream", "example", nil); err == nil || verror.ErrorID(err) != "v.io/x/ref/lib/pubsub.errNeedNonNilChannel" {
 		t.Fatalf("missing or wrong error: %v", err)
 	}
 	if _, err := pub.CreateStream("stream", "example", ch); err != nil {
 		t.Fatalf("unexpected error: %s", err)
 	}
 	if _, err := pub.CreateStream("stream", "example", ch); err == nil || verror.ErrorID(err) != "v.io/x/ref/lib/pubsub.errStreamExists" {
 		t.Fatalf("missing or wrong error: %v", err)
 	}
 	if got, want := pub.String(), "(stream: example)"; got != want {
 		t.Errorf("got %q, want %q", got, want)
 	}
 	stop, err := pub.CreateStream("s2", "eg2", ch)
 	if err != nil {
 		t.Fatalf("unexpected error: %s", err)
 	}
 	if got, want := pub.String(), "(stream: example) (s2: eg2)"; got != want {
 		wantAlternate := "(s2: eg2) (stream: example)"
 		if got != wantAlternate {
 			t.Errorf("got %q, want %q or %q", got, want, wantAlternate)
 		}
 	}

 	got, want := pub.Latest("s2"), &pubsub.Stream{
 		Name:        "s2",
 		Description: "eg2",
 		Latest:      nil,
 	}
 	if !reflect.DeepEqual(got, want) {
 		t.Errorf("got %q, want %q", got, want)
 	}
 	pub.Shutdown()
 	if _, running := <-stop; running {
 		t.Errorf("expected to be shutting down")
 	}
 	if _, err := pub.ForkStream("stream", nil); err == nil || verror.ErrorID(err) != "v.io/x/ref/lib/pubsub.errStreamShutDown" {
 		t.Errorf("missing or wrong error: %v", err)
 	}
 	if got, want := pub.String(), "shutdown"; got != want {
 		t.Errorf("got %s, want %s", got, want)
 	}

 }

 func producer(pub *pubsub.Publisher, in chan<- pubsub.Setting, stop <-chan struct{}, limit int, ch chan int, waiter *sync.WaitGroup) {
 	for i := 0; ; i++ {
 		select {
 		case <-stop:
 			ch <- i
 			waiter.Done()
 			// must close this channel, otherwise the Publisher will leak a goroutine for this stream.
 			close(in)
 			return
 		default:
 			// signal progress on ch, at limit/2, limit and when we're done (above)
 			switch {
 			case i == limit/2:
 				ch <- i
 			case i == limit:
 				ch <- i
 			}
 			if i%2 == 0 {
 				in <- pubsub.NewInt("i", "int", i)
 			} else {
 				in <- pubsub.NewFloat64("f", "float", float64(i))
 			}
 		}
 	}
 }

 func consumer(t *testing.T, pub *pubsub.Publisher, limit, bufsize int, errch chan error, starter, waiter *sync.WaitGroup) {
 	defer close(errch)
 	ch := make(chan pubsub.Setting, bufsize)
 	st, err := pub.ForkStream("net", ch)
 	if err != nil {
 		errch <- err
 		return
 	}
 	starter.Done()
 	i, i2 := 0, 0
 	if st.Latest["i"] != nil {
 		i = int(st.Latest["i"].Value().(int))
 	}
 	if st.Latest["f"] != nil {
 		i2 = int(st.Latest["f"].Value().(float64))
 	}
 	if i2 > i {
 		i = i2
 	}
 	i++
 	for s := range ch {
 		switch v := s.Value().(type) {
 		case int:
 			if i%2 != 0 {
 				errch <- fmt.Errorf("expected a float, got an int")
 				return
 			}
 			if v != i {
 				errch <- fmt.Errorf("got %d, want %d", v, i)
 				return
 			}
 		case float64:
 			if i%2 != 1 {
 				errch <- fmt.Errorf("expected an int, got a float")
 				return
 			}
 			if v != float64(i) {
 				errch <- fmt.Errorf("got %f, want %f", v, float64(i))
 				return
 			}
 		}
 		i++
 	}
 	if i < limit {
 		errch <- fmt.Errorf("didn't read enough settings: got %d, want >= %d", i, limit)
 		return
 	}
 	waiter.Done()
 }

 func testStream(t *testing.T, consumerBufSize int) {
 	in := make(chan pubsub.Setting)
 	pub := pubsub.NewPublisher()
 	stop, err := pub.CreateStream("net", "network settings", in)
 	if err != nil {
 		t.Fatal(err)
 	}

 	rand.Seed(time.Now().UnixNano())
 	limit := rand.Intn(5000)
 	if limit < 100 {
 		limit = 100
 	}
 	t.Logf("limit: %d", limit)

 	var waiter sync.WaitGroup
 	waiter.Add(3)

 	progress := make(chan int)
 	go producer(pub, in, stop, limit, progress, &waiter)

 	i := <-progress
 	t.Logf("limit/2 = %d", i)

 	err1 := make(chan error, 1)
 	err2 := make(chan error, 1)
 	var starter sync.WaitGroup
 	starter.Add(2)
 	go consumer(t, pub, limit, consumerBufSize, err1, &starter, &waiter)
 	go consumer(t, pub, limit, consumerBufSize, err2, &starter, &waiter)

 	reached := <-progress
 	// Give the consumers a chance to get going before shutting down
 	// the producer.
 	starter.Wait()
 	time.Sleep(100 * time.Millisecond)
 	pub.Shutdown()
 	shutdown := <-progress
 	t.Logf("reached %d, shut down at %d", reached, shutdown)

 	// This is a little annoying, we check for the presence of errors on the error
 	// channels once everything has run its course  since it's not allowed to call
 	// t.Fatal/Errorf from a separate goroutine. We wait until here so that we
 	// don't block waiting for errors that don't occur when the tests all work.
 	err = <-err1
 	if err != nil {
 		t.Fatal(err)
 	}
 	err = <-err2
 	if err != nil {
 		t.Fatal(err)
 	}
 	// Wait for all goroutines to finish.
 	waiter.Wait()
 }

 func TestStream(t *testing.T) {
 	testStream(t, 500)
 }

 func TestStreamSmallBuffers(t *testing.T) {
 	testStream(t, 1)
 }

 func TestDurationFlag(t *testing.T) {
 	d := &pubsub.DurationFlag{}
 	if err := d.Set("1s"); err != nil {
 		t.Errorf("unexpected error %s", err)
 	}
 	if got, want := d.Duration, time.Duration(time.Second); got != want {
 		t.Errorf("got %s, expected %s", got, want)
 	}
 	if err := d.Set("1t"); err == nil || err.Error() != "time: unknown unit t in duration 1t" {
 		t.Errorf("expected error %v", err)
 	}
 }
	// 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 pubsub_test

	import (
	"fmt"
	"math/rand"
	"reflect"
	"sync"
	"testing"
	"time"

	"v.io/x/ref/lib/pubsub"

	"v.io/v23/verror"
	)

	func ExamplePublisher() {
	in := make(chan pubsub.Setting)
	pub := pubsub.NewPublisher()
	pub.CreateStream("net", "network settings", in)

	// A simple producer of IP address settings.
	producer := func() {
	in <- pubsub.NewString("ip", "address", "1.2.3.5")
	}

	var waiter sync.WaitGroup
	waiter.Add(2)

	// A simple consumer of IP address Settings.
	consumer := func(ch chan pubsub.Setting) {
	fmt.Println(<-ch)
	waiter.Done()
	}

	// Publish an initial Setting to the Stream.
	in <- pubsub.NewString("ip", "address", "1.2.3.4")

	// Fork the stream twice, and read the latest value.
	ch1 := make(chan pubsub.Setting)
	st, _ := pub.ForkStream("net", ch1)
	fmt.Println(st.Latest["ip"])
	ch2 := make(chan pubsub.Setting)
	st, _ = pub.ForkStream("net", ch2)
	fmt.Println(st.Latest["ip"])

	// Now we can read new Settings as they are generated.
	go producer()
	go consumer(ch1)
	go consumer(ch2)

	waiter.Wait()

	// Output:
	// ip: address: (string: 1.2.3.4)
	// ip: address: (string: 1.2.3.4)
	// ip: address: (string: 1.2.3.5)
	// ip: address: (string: 1.2.3.5)
	}

	func ExampleShutdown() {
	in := make(chan pubsub.Setting)
	pub := pubsub.NewPublisher()
	stop, _ := pub.CreateStream("net", "network settings", in)

	var producerReady sync.WaitGroup
	producerReady.Add(1)
	var consumersReady sync.WaitGroup
	consumersReady.Add(2)

	// A producer to write 100 Settings before signalling that it's
	// ready to be shutdown. This is purely to demonstrate how to use
	// Shutdown.
	producer := func() {
	for i := 0; ; i++ {
	select {
	case <-stop:
	close(in)
	return
	default:
	in <- pubsub.NewString("ip", "address", "1.2.3.4")
	if i == 100 {
	producerReady.Done()
	}
	}
	}
	}

	var waiter sync.WaitGroup
	waiter.Add(2)

	consumer := func() {
	ch := make(chan pubsub.Setting, 10)
	pub.ForkStream("net", ch)
	consumersReady.Done()
	i := 0
	for {
	if _, ok := <-ch; !ok {
	// The channel has been closed when the publisher
	// is asked to shut down.
	break
	}
	i++
	}
	if i >= 100 {
	// We've received at least 100 Settings as per the producer above.
	fmt.Println("done")
	}
	waiter.Done()
	}

	go consumer()
	go consumer()
	consumersReady.Wait()
	go producer()
	producerReady.Wait()
	pub.Shutdown()
	waiter.Wait()
	// Output:
	// done
	// done
	}

	func TestSimple(t *testing.T) {
	ch := make(chan pubsub.Setting, 2)
	pub := pubsub.NewPublisher()
	if _, err := pub.ForkStream("stream", nil); err == nil \|\| verror.ErrorID(err) != "v.io/x/ref/lib/pubsub.errStreamDoesntExist" {
	t.Errorf("missing or wrong error: %v", err)
	}
	if _, err := pub.CreateStream("stream", "example", nil); err == nil \|\| verror.ErrorID(err) != "v.io/x/ref/lib/pubsub.errNeedNonNilChannel" {
	t.Fatalf("missing or wrong error: %v", err)
	}
	if _, err := pub.CreateStream("stream", "example", ch); err != nil {
	t.Fatalf("unexpected error: %s", err)
	}
	if _, err := pub.CreateStream("stream", "example", ch); err == nil \|\| verror.ErrorID(err) != "v.io/x/ref/lib/pubsub.errStreamExists" {
	t.Fatalf("missing or wrong error: %v", err)
	}
	if got, want := pub.String(), "(stream: example)"; got != want {
	t.Errorf("got %q, want %q", got, want)
	}
	stop, err := pub.CreateStream("s2", "eg2", ch)
	if err != nil {
	t.Fatalf("unexpected error: %s", err)
	}
	if got, want := pub.String(), "(stream: example) (s2: eg2)"; got != want {
	wantAlternate := "(s2: eg2) (stream: example)"
	if got != wantAlternate {
	t.Errorf("got %q, want %q or %q", got, want, wantAlternate)
	}
	}

	got, want := pub.Latest("s2"), &pubsub.Stream{
	Name: "s2",
	Description: "eg2",
	Latest: nil,
	}
	if !reflect.DeepEqual(got, want) {
	t.Errorf("got %q, want %q", got, want)
	}
	pub.Shutdown()
	if _, running := <-stop; running {
	t.Errorf("expected to be shutting down")
	}
	if _, err := pub.ForkStream("stream", nil); err == nil \|\| verror.ErrorID(err) != "v.io/x/ref/lib/pubsub.errStreamShutDown" {
	t.Errorf("missing or wrong error: %v", err)
	}
	if got, want := pub.String(), "shutdown"; got != want {
	t.Errorf("got %s, want %s", got, want)
	}

	}

	func producer(pub pubsub.Publisher, in chan<- pubsub.Setting, stop <-chan struct{}, limit int, ch chan int, waiter sync.WaitGroup) {
	for i := 0; ; i++ {
	select {
	case <-stop:
	ch <- i
	waiter.Done()
	// must close this channel, otherwise the Publisher will leak a goroutine for this stream.
	close(in)
	return
	default:
	// signal progress on ch, at limit/2, limit and when we're done (above)
	switch {
	case i == limit/2:
	ch <- i
	case i == limit:
	ch <- i
	}
	if i%2 == 0 {
	in <- pubsub.NewInt("i", "int", i)
	} else {
	in <- pubsub.NewFloat64("f", "float", float64(i))
	}
	}
	}
	}

	func consumer(t testing.T, pub pubsub.Publisher, limit, bufsize int, errch chan error, starter, waiter *sync.WaitGroup) {
	defer close(errch)
	ch := make(chan pubsub.Setting, bufsize)
	st, err := pub.ForkStream("net", ch)
	if err != nil {
	errch <- err
	return
	}
	starter.Done()
	i, i2 := 0, 0
	if st.Latest["i"] != nil {
	i = int(st.Latest["i"].Value().(int))
	}
	if st.Latest["f"] != nil {
	i2 = int(st.Latest["f"].Value().(float64))
	}
	if i2 > i {
	i = i2
	}
	i++
	for s := range ch {
	switch v := s.Value().(type) {
	case int:
	if i%2 != 0 {
	errch <- fmt.Errorf("expected a float, got an int")
	return
	}
	if v != i {
	errch <- fmt.Errorf("got %d, want %d", v, i)
	return
	}
	case float64:
	if i%2 != 1 {
	errch <- fmt.Errorf("expected an int, got a float")
	return
	}
	if v != float64(i) {
	errch <- fmt.Errorf("got %f, want %f", v, float64(i))
	return
	}
	}
	i++
	}
	if i < limit {
	errch <- fmt.Errorf("didn't read enough settings: got %d, want >= %d", i, limit)
	return
	}
	waiter.Done()
	}

	func testStream(t *testing.T, consumerBufSize int) {
	in := make(chan pubsub.Setting)
	pub := pubsub.NewPublisher()
	stop, err := pub.CreateStream("net", "network settings", in)
	if err != nil {
	t.Fatal(err)
	}

	rand.Seed(time.Now().UnixNano())
	limit := rand.Intn(5000)
	if limit < 100 {
	limit = 100
	}
	t.Logf("limit: %d", limit)

	var waiter sync.WaitGroup
	waiter.Add(3)

	progress := make(chan int)
	go producer(pub, in, stop, limit, progress, &waiter)

	i := <-progress
	t.Logf("limit/2 = %d", i)

	err1 := make(chan error, 1)
	err2 := make(chan error, 1)
	var starter sync.WaitGroup
	starter.Add(2)
	go consumer(t, pub, limit, consumerBufSize, err1, &starter, &waiter)
	go consumer(t, pub, limit, consumerBufSize, err2, &starter, &waiter)

	reached := <-progress
	// Give the consumers a chance to get going before shutting down
	// the producer.
	starter.Wait()
	time.Sleep(100 * time.Millisecond)
	pub.Shutdown()
	shutdown := <-progress
	t.Logf("reached %d, shut down at %d", reached, shutdown)

	// This is a little annoying, we check for the presence of errors on the error
	// channels once everything has run its course since it's not allowed to call
	// t.Fatal/Errorf from a separate goroutine. We wait until here so that we
	// don't block waiting for errors that don't occur when the tests all work.
	err = <-err1
	if err != nil {
	t.Fatal(err)
	}
	err = <-err2
	if err != nil {
	t.Fatal(err)
	}
	// Wait for all goroutines to finish.
	waiter.Wait()
	}

	func TestStream(t *testing.T) {
	testStream(t, 500)
	}

	func TestStreamSmallBuffers(t *testing.T) {
	testStream(t, 1)
	}

	func TestDurationFlag(t *testing.T) {
	d := &pubsub.DurationFlag{}
	if err := d.Set("1s"); err != nil {
	t.Errorf("unexpected error %s", err)
	}
	if got, want := d.Duration, time.Duration(time.Second); got != want {
	t.Errorf("got %s, expected %s", got, want)
	}
	if err := d.Set("1t"); err == nil \|\| err.Error() != "time: unknown unit t in duration 1t" {
	t.Errorf("expected error %v", err)
	}
	}