go/pkg/electron/electron_test.go - qpid-proton - Git at Google

 /*
 Licensed to the Apache Software Foundation (ASF) under one
 or more contributor license agreements.  See the NOTICE file
 distributed with this work for additional information
 regarding copyright ownership.  The ASF licenses this file
 to you under the Apache License, Version 2.0 (the
 "License"); you may not use this file except in compliance
 with the License.  You may obtain a copy of the License at

   http://www.apache.org/licenses/LICENSE-2.0

 Unless required by applicable law or agreed to in writing,
 software distributed under the License is distributed on an
 "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 KIND, either express or implied.  See the License for the
 specific language governing permissions and limitations
 under the License.
 */

 package electron

 import (
 	"fmt"
 	"testing"
 	"time"

 	"github.com/apache/qpid-proton/go/pkg/amqp"
 	"github.com/apache/qpid-proton/go/pkg/internal/test"
 )

 // Send a message one way with a client sender and server receiver, verify ack.
 func TestClientSender(t *testing.T) {
 	p := newPipe(t, nil, nil)
 	defer func() { p.close() }()

 	nLinks := 3
 	nMessages := 3

 	s := make([]Sender, nLinks)
 	r := make([]Receiver, nLinks)
 	for i := 0; i < nLinks; i++ {
 		s[i], r[i] = p.sender(Target(fmt.Sprintf("foo%d", i)))
 	}

 	for i := 0; i < nLinks; i++ {
 		for j := 0; j < nMessages; j++ {
 			// Client send
 			ack := make(chan Outcome, 1)
 			sendDone := make(chan struct{})
 			go func() {
 				defer close(sendDone)
 				m := amqp.NewMessageWith(fmt.Sprintf("foobar%v-%v", i, j))
 				var err error
 				s[i].SendAsync(m, ack, "testing")
 				test.FatalIf(t, err)
 			}()

 			// Server receive
 			rm, err := r[i].Receive()
 			test.FatalIf(t, err)
 			if want, got := interface{}(fmt.Sprintf("foobar%v-%v", i, j)), rm.Message.Body(); want != got {
 				t.Errorf("%#v != %#v", want, got)
 			}

 			// Should not be acknowledged on client yet
 			<-sendDone
 			select {
 			case <-ack:
 				t.Errorf("unexpected ack")
 			default:
 			}

 			// Server send ack
 			if err = rm.Reject(); err != nil {
 				t.Error(err)
 			}
 			// Client get ack.
 			if a := <-ack; a.Value != "testing" || a.Error != nil || a.Status != Rejected {
 				t.Error("unexpected ack: ", a.Status, a.Error, a.Value)
 			}
 		}
 	}
 }

 func TestClientReceiver(t *testing.T) {
 	nMessages := 3
 	p := newPipe(t, nil, nil)
 	defer func() { p.close() }()
 	r, s := p.receiver(Source("foo"), Capacity(nMessages), Prefetch(true))
 	go func() {
 		for i := 0; i < nMessages; i++ { // Server sends
 			out := s.SendSync(amqp.NewMessageWith(int32(i)))
 			test.FatalIf(t, out.Error)
 		}
 	}()
 	for i := 0; i < nMessages; i++ { // Client receives
 		rm, err := r.Receive()
 		test.FatalIf(t, err)
 		test.ErrorIf(t, test.Differ(int32(i), rm.Message.Body()))
 		test.ErrorIf(t, rm.Accept())
 	}
 }

 // Test timeout versions of waiting functions.
 func TestTimeouts(t *testing.T) {
 	p := newPipe(t, nil, nil)
 	defer func() { p.close() }()
 	snd, rcv := p.sender(Target("test"))

 	// Test send with timeout
 	short := time.Millisecond
 	long := time.Second
 	m := amqp.NewMessage()
 	if err := snd.SendSyncTimeout(m, 0).Error; err != Timeout { // No credit, expect timeout.
 		t.Error("want Timeout got", err)
 	}
 	if err := snd.SendSyncTimeout(m, short).Error; err != Timeout { // No credit, expect timeout.
 		t.Error("want Timeout got", err)
 	}
 	// Test receive with timeout
 	if _, err := rcv.ReceiveTimeout(0); err != Timeout { // No credit, expect timeout.
 		t.Error("want Timeout got", err)
 	}
 	// Test receive with timeout
 	if _, err := rcv.ReceiveTimeout(short); err != Timeout { // No credit, expect timeout.
 		t.Error("want Timeout got", err)
 	}
 	// There is now a credit on the link due to receive
 	ack := make(chan Outcome)
 	snd.SendAsyncTimeout(m, ack, nil, short)
 	// Disposition should timeout
 	select {
 	case <-ack:
 		t.Errorf("want Timeout got %#v", ack)
 	case <-time.After(short):
 	}

 	// Receive and accept
 	rm, err := rcv.ReceiveTimeout(long)
 	if err != nil {
 		t.Fatal(err)
 	}
 	if err = rm.Accept(); err != nil {
 		t.Fatal(err)
 	}
 	// Sender get ack
 	if a := <-ack; a.Status != Accepted || a.Error != nil {
 		t.Errorf("want (accepted, nil) got %#v", a)
 	}
 }

 type result struct {
 	label string
 	err   error
 	value interface{}
 }

 func (r result) String() string { return fmt.Sprintf("%v(%v)", r.err, r.label) }

 func doSend(snd Sender, results chan result) {
 	err := snd.SendSync(amqp.NewMessage()).Error
 	results <- result{"send", err, nil}
 }

 func doReceive(rcv Receiver, results chan result) {
 	msg, err := rcv.Receive()
 	results <- result{"receive", err, msg}
 }

 func doDisposition(ack <-chan Outcome, results chan result) {
 	results <- result{"disposition", (<-ack).Error, nil}
 }

 // Senders get credit immediately if receivers have prefetch set
 func TestSendReceivePrefetch(t *testing.T) {
 	p := newPipe(t, nil, nil)
 	p.prefetch = true
 	s, r := p.sender()
 	s.SendAsyncTimeout(amqp.NewMessage(), nil, nil, time.Second) // Should not block for credit.
 	if _, err := r.Receive(); err != nil {
 		t.Error(err)
 	}
 }

 // Senders do not get credit till Receive() if receivers don't have prefetch
 func TestSendReceiveNoPrefetch(t *testing.T) {
 	p := newPipe(t, nil, nil)
 	p.prefetch = false
 	s, r := p.sender()
 	done := make(chan struct{}, 1)
 	go func() {
 		s.SendAsyncTimeout(amqp.NewMessage(), nil, nil, time.Second) // Should block for credit.
 		close(done)
 	}()
 	select {
 	case <-done:
 		t.Errorf("send should be blocked on credit")
 	default:
 		if _, err := r.Receive(); err != nil {
 			t.Error(err)
 		} else {
 			<-done
 		} // Should be unblocked now
 	}
 }

 // Test that closing Links interrupts blocked link functions.
 func TestLinkCloseInterrupt(t *testing.T) {
 	want := amqp.Error{Name: "x", Description: "all bad"}
 	p := newPipe(t, nil, nil)
 	results := make(chan result) // Collect expected errors

 	// Note closing the link does not interrupt Send() calls, the AMQP spec says
 	// that deliveries can be settled after the link is closed.

 	// Receiver.Close() interrupts Receive()
 	snd, rcv := p.sender()
 	go doReceive(rcv, results)
 	rcv.Close(want)
 	if r := <-results; want != r.err {
 		t.Errorf("want %#v got %#v", want, r)
 	}

 	// Remote Sender.Close() interrupts Receive()
 	snd, rcv = p.sender()
 	go doReceive(rcv, results)
 	snd.Close(want)
 	if r := <-results; want != r.err {
 		t.Errorf("want %#v got %#v", want, r)
 	}
 }

 // Test closing the server end of a connection.
 func TestConnectionCloseInterrupt1(t *testing.T) {
 	want := amqp.Error{Name: "x", Description: "bad"}
 	p := newSocketPair(t, nil, nil)
 	p.prefetch = true
 	results := make(chan result) // Collect expected errors

 	// Connection.Close() interrupts Send, Receive, Disposition.
 	snd, rcv := p.sender()
 	go doSend(snd, results)

 	if _, err := rcv.Receive(); err != nil {
 		t.Error("receive", err)
 	}
 	rcv, snd = p.receiver()
 	go doReceive(rcv, results)

 	snd, rcv = p.sender()
 	ack := snd.SendWaitable(amqp.NewMessage())
 	if _, err := rcv.Receive(); err != nil {
 		t.Error("receive", err)
 	}
 	go doDisposition(ack, results)

 	p.server.Close(want)
 	for i := 0; i < 3; i++ {
 		if r := <-results; want != r.err {
 			t.Errorf("want %v got %v", want, r)
 		}
 	}
 }

 // Test closing the client end of the connection.
 func TestConnectionCloseInterrupt2(t *testing.T) {
 	want := amqp.Error{Name: "x", Description: "bad"}
 	p := newSocketPair(t, nil, nil)
 	p.prefetch = true
 	results := make(chan result) // Collect expected errors

 	// Connection.Close() interrupts Send, Receive, Disposition.
 	snd, rcv := p.sender()
 	go doSend(snd, results)
 	if _, err := rcv.Receive(); err != nil {
 		t.Error("receive", err)
 	}

 	rcv, snd = p.receiver()
 	go doReceive(rcv, results)

 	snd, rcv = p.sender()
 	ack := snd.SendWaitable(amqp.NewMessage())
 	go doDisposition(ack, results)

 	p.client.Connection().Close(want)
 	for i := 0; i < 3; i++ {
 		if r := <-results; want != r.err {
 			t.Errorf("want %v got %v", want, r.err)
 		}
 	}
 }

 func TestHeartbeat(t *testing.T) {
 	p := newSocketPair(t,
 		[]ConnectionOption{Heartbeat(102 * time.Millisecond)},
 		[]ConnectionOption{Heartbeat(101 * time.Millisecond)})
 	defer func() { p.close() }()

 	// Function to freeze the server to stop it sending heartbeats.
 	unfreeze := make(chan bool)
 	defer close(unfreeze)
 	freeze := func() error { return p.server.(*connection).engine.Inject(func() { <-unfreeze }) }

 	test.FatalIf(t, p.client.Sync())
 	test.ErrorIf(t, test.Differ(101*time.Millisecond, p.client.Connection().Heartbeat()))
 	test.ErrorIf(t, test.Differ(102*time.Millisecond, p.server.Heartbeat()))

 	// Freeze the server for less than a heartbeat
 	test.FatalIf(t, freeze())
 	time.Sleep(5 * time.Millisecond)
 	unfreeze <- true
 	// Make sure server is still responding.
 	s, _ := p.sender()
 	test.ErrorIf(t, s.Sync())

 	// Freeze the server till the p.client times out the connection
 	test.FatalIf(t, freeze())
 	select {
 	case <-p.client.Done():
 		if amqp.ResourceLimitExceeded != p.client.Error().(amqp.Error).Name {
 			t.Error("bad timeout error:", p.client.Error())
 		}
 	case <-time.After(1400 * time.Millisecond):
 		t.Error("connection failed to time out")
 	}

 	unfreeze <- true // Unfreeze the server
 	<-p.server.Done()
 	if p.server.Error() == nil {
 		t.Error("expected server side  time-out or connection abort error")
 	}
 }
	/*
	Licensed to the Apache Software Foundation (ASF) under one
	or more contributor license agreements. See the NOTICE file
	distributed with this work for additional information
	regarding copyright ownership. The ASF licenses this file
	to you under the Apache License, Version 2.0 (the
	"License"); you may not use this file except in compliance
	with the License. You may obtain a copy of the License at

	http://www.apache.org/licenses/LICENSE-2.0

	Unless required by applicable law or agreed to in writing,
	software distributed under the License is distributed on an
	"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	KIND, either express or implied. See the License for the
	specific language governing permissions and limitations
	under the License.
	*/

	package electron

	import (
	"fmt"
	"testing"
	"time"

	"github.com/apache/qpid-proton/go/pkg/amqp"
	"github.com/apache/qpid-proton/go/pkg/internal/test"
	)

	// Send a message one way with a client sender and server receiver, verify ack.
	func TestClientSender(t *testing.T) {
	p := newPipe(t, nil, nil)
	defer func() { p.close() }()

	nLinks := 3
	nMessages := 3

	s := make([]Sender, nLinks)
	r := make([]Receiver, nLinks)
	for i := 0; i < nLinks; i++ {
	s[i], r[i] = p.sender(Target(fmt.Sprintf("foo%d", i)))
	}

	for i := 0; i < nLinks; i++ {
	for j := 0; j < nMessages; j++ {
	// Client send
	ack := make(chan Outcome, 1)
	sendDone := make(chan struct{})
	go func() {
	defer close(sendDone)
	m := amqp.NewMessageWith(fmt.Sprintf("foobar%v-%v", i, j))
	var err error
	s[i].SendAsync(m, ack, "testing")
	test.FatalIf(t, err)
	}()

	// Server receive
	rm, err := r[i].Receive()
	test.FatalIf(t, err)
	if want, got := interface{}(fmt.Sprintf("foobar%v-%v", i, j)), rm.Message.Body(); want != got {
	t.Errorf("%#v != %#v", want, got)
	}

	// Should not be acknowledged on client yet
	<-sendDone
	select {
	case <-ack:
	t.Errorf("unexpected ack")
	default:
	}

	// Server send ack
	if err = rm.Reject(); err != nil {
	t.Error(err)
	}
	// Client get ack.
	if a := <-ack; a.Value != "testing" \|\| a.Error != nil \|\| a.Status != Rejected {
	t.Error("unexpected ack: ", a.Status, a.Error, a.Value)
	}
	}
	}
	}

	func TestClientReceiver(t *testing.T) {
	nMessages := 3
	p := newPipe(t, nil, nil)
	defer func() { p.close() }()
	r, s := p.receiver(Source("foo"), Capacity(nMessages), Prefetch(true))
	go func() {
	for i := 0; i < nMessages; i++ { // Server sends
	out := s.SendSync(amqp.NewMessageWith(int32(i)))
	test.FatalIf(t, out.Error)
	}
	}()
	for i := 0; i < nMessages; i++ { // Client receives
	rm, err := r.Receive()
	test.FatalIf(t, err)
	test.ErrorIf(t, test.Differ(int32(i), rm.Message.Body()))
	test.ErrorIf(t, rm.Accept())
	}
	}

	// Test timeout versions of waiting functions.
	func TestTimeouts(t *testing.T) {
	p := newPipe(t, nil, nil)
	defer func() { p.close() }()
	snd, rcv := p.sender(Target("test"))

	// Test send with timeout
	short := time.Millisecond
	long := time.Second
	m := amqp.NewMessage()
	if err := snd.SendSyncTimeout(m, 0).Error; err != Timeout { // No credit, expect timeout.
	t.Error("want Timeout got", err)
	}
	if err := snd.SendSyncTimeout(m, short).Error; err != Timeout { // No credit, expect timeout.
	t.Error("want Timeout got", err)
	}
	// Test receive with timeout
	if _, err := rcv.ReceiveTimeout(0); err != Timeout { // No credit, expect timeout.
	t.Error("want Timeout got", err)
	}
	// Test receive with timeout
	if _, err := rcv.ReceiveTimeout(short); err != Timeout { // No credit, expect timeout.
	t.Error("want Timeout got", err)
	}
	// There is now a credit on the link due to receive
	ack := make(chan Outcome)
	snd.SendAsyncTimeout(m, ack, nil, short)
	// Disposition should timeout
	select {
	case <-ack:
	t.Errorf("want Timeout got %#v", ack)
	case <-time.After(short):
	}

	// Receive and accept
	rm, err := rcv.ReceiveTimeout(long)
	if err != nil {
	t.Fatal(err)
	}
	if err = rm.Accept(); err != nil {
	t.Fatal(err)
	}
	// Sender get ack
	if a := <-ack; a.Status != Accepted \|\| a.Error != nil {
	t.Errorf("want (accepted, nil) got %#v", a)
	}
	}

	type result struct {
	label string
	err error
	value interface{}
	}

	func (r result) String() string { return fmt.Sprintf("%v(%v)", r.err, r.label) }

	func doSend(snd Sender, results chan result) {
	err := snd.SendSync(amqp.NewMessage()).Error
	results <- result{"send", err, nil}
	}

	func doReceive(rcv Receiver, results chan result) {
	msg, err := rcv.Receive()
	results <- result{"receive", err, msg}
	}

	func doDisposition(ack <-chan Outcome, results chan result) {
	results <- result{"disposition", (<-ack).Error, nil}
	}

	// Senders get credit immediately if receivers have prefetch set
	func TestSendReceivePrefetch(t *testing.T) {
	p := newPipe(t, nil, nil)
	p.prefetch = true
	s, r := p.sender()
	s.SendAsyncTimeout(amqp.NewMessage(), nil, nil, time.Second) // Should not block for credit.
	if _, err := r.Receive(); err != nil {
	t.Error(err)
	}
	}

	// Senders do not get credit till Receive() if receivers don't have prefetch
	func TestSendReceiveNoPrefetch(t *testing.T) {
	p := newPipe(t, nil, nil)
	p.prefetch = false
	s, r := p.sender()
	done := make(chan struct{}, 1)
	go func() {
	s.SendAsyncTimeout(amqp.NewMessage(), nil, nil, time.Second) // Should block for credit.
	close(done)
	}()
	select {
	case <-done:
	t.Errorf("send should be blocked on credit")
	default:
	if _, err := r.Receive(); err != nil {
	t.Error(err)
	} else {
	<-done
	} // Should be unblocked now
	}
	}

	// Test that closing Links interrupts blocked link functions.
	func TestLinkCloseInterrupt(t *testing.T) {
	want := amqp.Error{Name: "x", Description: "all bad"}
	p := newPipe(t, nil, nil)
	results := make(chan result) // Collect expected errors

	// Note closing the link does not interrupt Send() calls, the AMQP spec says
	// that deliveries can be settled after the link is closed.

	// Receiver.Close() interrupts Receive()
	snd, rcv := p.sender()
	go doReceive(rcv, results)
	rcv.Close(want)
	if r := <-results; want != r.err {
	t.Errorf("want %#v got %#v", want, r)
	}

	// Remote Sender.Close() interrupts Receive()
	snd, rcv = p.sender()
	go doReceive(rcv, results)
	snd.Close(want)
	if r := <-results; want != r.err {
	t.Errorf("want %#v got %#v", want, r)
	}
	}

	// Test closing the server end of a connection.
	func TestConnectionCloseInterrupt1(t *testing.T) {
	want := amqp.Error{Name: "x", Description: "bad"}
	p := newSocketPair(t, nil, nil)
	p.prefetch = true
	results := make(chan result) // Collect expected errors

	// Connection.Close() interrupts Send, Receive, Disposition.
	snd, rcv := p.sender()
	go doSend(snd, results)

	if _, err := rcv.Receive(); err != nil {
	t.Error("receive", err)
	}
	rcv, snd = p.receiver()
	go doReceive(rcv, results)

	snd, rcv = p.sender()
	ack := snd.SendWaitable(amqp.NewMessage())
	if _, err := rcv.Receive(); err != nil {
	t.Error("receive", err)
	}
	go doDisposition(ack, results)

	p.server.Close(want)
	for i := 0; i < 3; i++ {
	if r := <-results; want != r.err {
	t.Errorf("want %v got %v", want, r)
	}
	}
	}

	// Test closing the client end of the connection.
	func TestConnectionCloseInterrupt2(t *testing.T) {
	want := amqp.Error{Name: "x", Description: "bad"}
	p := newSocketPair(t, nil, nil)
	p.prefetch = true
	results := make(chan result) // Collect expected errors

	// Connection.Close() interrupts Send, Receive, Disposition.
	snd, rcv := p.sender()
	go doSend(snd, results)
	if _, err := rcv.Receive(); err != nil {
	t.Error("receive", err)
	}

	rcv, snd = p.receiver()
	go doReceive(rcv, results)

	snd, rcv = p.sender()
	ack := snd.SendWaitable(amqp.NewMessage())
	go doDisposition(ack, results)

	p.client.Connection().Close(want)
	for i := 0; i < 3; i++ {
	if r := <-results; want != r.err {
	t.Errorf("want %v got %v", want, r.err)
	}
	}
	}

	func TestHeartbeat(t *testing.T) {
	p := newSocketPair(t,
	[]ConnectionOption{Heartbeat(102 * time.Millisecond)},
	[]ConnectionOption{Heartbeat(101 * time.Millisecond)})
	defer func() { p.close() }()

	// Function to freeze the server to stop it sending heartbeats.
	unfreeze := make(chan bool)
	defer close(unfreeze)
	freeze := func() error { return p.server.(*connection).engine.Inject(func() { <-unfreeze }) }

	test.FatalIf(t, p.client.Sync())
	test.ErrorIf(t, test.Differ(101*time.Millisecond, p.client.Connection().Heartbeat()))
	test.ErrorIf(t, test.Differ(102*time.Millisecond, p.server.Heartbeat()))

	// Freeze the server for less than a heartbeat
	test.FatalIf(t, freeze())
	time.Sleep(5 * time.Millisecond)
	unfreeze <- true
	// Make sure server is still responding.
	s, _ := p.sender()
	test.ErrorIf(t, s.Sync())

	// Freeze the server till the p.client times out the connection
	test.FatalIf(t, freeze())
	select {
	case <-p.client.Done():
	if amqp.ResourceLimitExceeded != p.client.Error().(amqp.Error).Name {
	t.Error("bad timeout error:", p.client.Error())
	}
	case <-time.After(1400 * time.Millisecond):
	t.Error("connection failed to time out")
	}

	unfreeze <- true // Unfreeze the server
	<-p.server.Done()
	if p.server.Error() == nil {
	t.Error("expected server side time-out or connection abort error")
	}
	}