go/pkg/electron/sender.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

 // #include <proton/disposition.h>
 import "C"

 import (
 	"fmt"
 	"time"

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

 // Sender is a Link that sends messages.
 //
 // The result of sending a message is provided by an Outcome value.
 //
 // A sender can buffer messages up to the credit limit provided by the remote receiver.
 // All the Send* methods will block if the buffer is full until there is space.
 // Send*Timeout methods will give up after the timeout and set Timeout as Outcome.Error.
 //
 type Sender interface {
 	Endpoint
 	LinkSettings

 	// SendSync sends a message and blocks until the message is acknowledged by the remote receiver.
 	// Returns an Outcome, which may contain an error if the message could not be sent.
 	SendSync(m amqp.Message) Outcome

 	// SendWaitable puts a message in the send buffer and returns a channel that
 	// you can use to wait for the Outcome of just that message. The channel is
 	// buffered so you can receive from it whenever you want without blocking.
 	//
 	// Note: can block if there is no space to buffer the message.
 	SendWaitable(m amqp.Message) <-chan Outcome

 	// SendForget buffers a message for sending and returns, with no notification of the outcome.
 	//
 	// Note: can block if there is no space to buffer the message.
 	SendForget(m amqp.Message)

 	// SendAsync puts a message in the send buffer and returns immediately.  An
 	// Outcome with Value = value will be sent to the ack channel when the remote
 	// receiver has acknowledged the message or if there is an error.
 	//
 	// You can use the same ack channel for many calls to SendAsync(), possibly on
 	// many Senders. The channel will receive the outcomes in the order they
 	// become available. The channel should be buffered and/or served by dedicated
 	// goroutines to avoid blocking the connection.
 	//
 	// If ack == nil no Outcome is sent.
 	//
 	// Note: can block if there is no space to buffer the message.
 	SendAsync(m amqp.Message, ack chan<- Outcome, value interface{})

 	SendAsyncTimeout(m amqp.Message, ack chan<- Outcome, value interface{}, timeout time.Duration)

 	SendWaitableTimeout(m amqp.Message, timeout time.Duration) <-chan Outcome

 	SendForgetTimeout(m amqp.Message, timeout time.Duration)

 	SendSyncTimeout(m amqp.Message, timeout time.Duration) Outcome
 }

 // Outcome provides information about the outcome of sending a message.
 type Outcome struct {
 	// Status of the message: was it sent, how was it acknowledged.
 	Status SentStatus
 	// Error is a local error if Status is Unsent or Unacknowledged, a remote error otherwise.
 	Error error
 	// Value provided by the application in SendAsync()
 	Value interface{}
 }

 func (o Outcome) send(ack chan<- Outcome) {
 	if ack != nil {
 		ack <- o
 	}
 }

 // SentStatus indicates the status of a sent message.
 type SentStatus int

 const (
 	// Message was never sent
 	Unsent SentStatus = iota
 	// Message was sent but never acknowledged. It may or may not have been received.
 	Unacknowledged
 	// Message was accepted by the receiver (or was sent pre-settled, accept is assumed)
 	Accepted
 	// Message was rejected as invalid by the receiver
 	Rejected
 	// Message was not processed by the receiver but may be valid for a different receiver
 	Released
 	// Receiver responded with an unrecognized status.
 	Unknown
 )

 // String human readable name for SentStatus.
 func (s SentStatus) String() string {
 	switch s {
 	case Unsent:
 		return "unsent"
 	case Unacknowledged:
 		return "unacknowledged"
 	case Accepted:
 		return "accepted"
 	case Rejected:
 		return "rejected"
 	case Released:
 		return "released"
 	case Unknown:
 		return "unknown"
 	default:
 		return fmt.Sprintf("invalid(%d)", s)
 	}
 }

 // Convert proton delivery state code to SentStatus value
 func sentStatus(d uint64) SentStatus {
 	switch d {
 	case proton.Accepted:
 		return Accepted
 	case proton.Rejected:
 		return Rejected
 	case proton.Released, proton.Modified:
 		return Released
 	default:
 		return Unknown
 	}
 }

 type sendable struct {
 	m    amqp.Message
 	ack  chan<- Outcome // Channel for acknowledgement of m
 	v    interface{}    // Correlation value
 	sent chan struct{}  // Closed when m is encoded and will be sent
 }

 func (sm *sendable) unsent(err error) {
 	Outcome{Unsent, err, sm.v}.send(sm.ack)
 }

 type sender struct {
 	link
 	sending []*sendable
 }

 func newSender(ls linkSettings) *sender {
 	s := &sender{link: link{linkSettings: ls}}
 	s.endpoint.init(s.link.pLink.String())
 	s.handler().addLink(s.pLink, s)
 	s.link.pLink.Open()
 	return s
 }

 // Called in handler goroutine
 func (s *sender) startSend(sm *sendable) {
 	s.sending = append(s.sending, sm)
 	s.trySend()
 }

 // Called in handler goroutine
 func (s *sender) trySend() {
 	for s.pLink.Credit() > 0 && len(s.sending) > 0 {
 		sm := s.sending[0]
 		s.sending = s.sending[1:]
 		s.send(sm)
 	}
 }

 // Called in handler goroutine with credit > 0
 func (s *sender) send(sm *sendable) {
 	if err := s.Error(); err != nil {
 		sm.unsent(err)
 		return
 	}
 	bytes, err := s.session.connection.mc.Encode(sm.m, nil)
 	close(sm.sent) // Safe to re-use sm.m now
 	if err != nil {
 		sm.unsent(err)
 		return
 	}
 	d, err := s.pLink.SendMessageBytes(bytes)
 	if err != nil {
 		sm.unsent(err)
 		return
 	}
 	if s.SndSettle() == SndSettled || (s.SndSettle() == SndMixed && sm.ack == nil) {
 		d.Settle()                                // Pre-settled
 		Outcome{Accepted, nil, sm.v}.send(sm.ack) // Assume accepted
 	} else {
 		// Register with handler to receive the remote outcome
 		s.handler().sent[d] = sm
 	}
 }

 func (s *sender) timeoutSend(sm *sendable) {
 	for i, sm2 := range s.sending {
 		if sm2 == sm {
 			n := copy(s.sending[i:], s.sending[i+1:])
 			s.sending = s.sending[:i+n] // delete
 			close(sm.sent)
 			return
 		}
 	}
 }

 func (s *sender) SendAsyncTimeout(m amqp.Message, ack chan<- Outcome, v interface{}, t time.Duration) {
 	sm := &sendable{m, ack, v, make(chan struct{})}
 	s.engine().Inject(func() { s.startSend(sm) })
 	select {
 	case <-sm.sent: // OK
 	case <-After(t): // Try to timeout sm
 		s.engine().Inject(func() { s.timeoutSend(sm) })
 	}
 }

 func (s *sender) SendWaitableTimeout(m amqp.Message, t time.Duration) <-chan Outcome {
 	out := make(chan Outcome, 1)
 	s.SendAsyncTimeout(m, out, nil, t)
 	return out
 }

 func (s *sender) SendForgetTimeout(m amqp.Message, t time.Duration) {
 	s.SendAsyncTimeout(m, nil, nil, t)
 }

 func (s *sender) SendSyncTimeout(m amqp.Message, t time.Duration) Outcome {
 	deadline := time.Now().Add(t)
 	ack := s.SendWaitableTimeout(m, t)
 	t = deadline.Sub(time.Now()) // Adjust for time already spent.
 	if t < 0 {
 		t = 0
 	}
 	if out, err := timedReceive(ack, t); err == nil {
 		return out.(Outcome)
 	} else {
 		if err == Closed && s.Error() != nil {
 			err = s.Error()
 		}
 		return Outcome{Unacknowledged, err, nil}
 	}
 }

 func (s *sender) SendAsync(m amqp.Message, ack chan<- Outcome, v interface{}) {
 	s.SendAsyncTimeout(m, ack, v, Forever)
 }

 func (s *sender) SendWaitable(m amqp.Message) <-chan Outcome {
 	return s.SendWaitableTimeout(m, Forever)
 }

 func (s *sender) SendForget(m amqp.Message) {
 	s.SendForgetTimeout(m, Forever)
 }

 func (s *sender) SendSync(m amqp.Message) Outcome {
 	return <-s.SendWaitable(m)
 }

 // handler goroutine
 func (s *sender) closed(err error) error {
 	for _, sm := range s.sending {
 		close(sm.sent)
 	}
 	s.sending = nil
 	return s.link.closed(err)
 }

 // IncomingSender is sent on the Connection.Incoming() channel when there is
 // an incoming request to open a sender link.
 type IncomingSender struct {
 	incoming
 	linkSettings
 }

 func newIncomingSender(sn *session, pLink proton.Link) *IncomingSender {
 	return &IncomingSender{
 		incoming:     makeIncoming(pLink),
 		linkSettings: makeIncomingLinkSettings(pLink, sn),
 	}
 }

 // Accept accepts an incoming sender endpoint
 func (in *IncomingSender) Accept() Endpoint {
 	return in.accept(func() Endpoint { return newSender(in.linkSettings) })
 }

 // Call in injected functions to check if the sender is valid.
 func (s *sender) valid() bool {
 	s2, ok := s.handler().links[s.pLink].(*sender)
 	return ok && s2 == s
 }
	/*
	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

	// #include <proton/disposition.h>
	import "C"

	import (
	"fmt"
	"time"

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

	// Sender is a Link that sends messages.
	//
	// The result of sending a message is provided by an Outcome value.
	//
	// A sender can buffer messages up to the credit limit provided by the remote receiver.
	// All the Send* methods will block if the buffer is full until there is space.
	// Send*Timeout methods will give up after the timeout and set Timeout as Outcome.Error.
	//
	type Sender interface {
	Endpoint
	LinkSettings

	// SendSync sends a message and blocks until the message is acknowledged by the remote receiver.
	// Returns an Outcome, which may contain an error if the message could not be sent.
	SendSync(m amqp.Message) Outcome

	// SendWaitable puts a message in the send buffer and returns a channel that
	// you can use to wait for the Outcome of just that message. The channel is
	// buffered so you can receive from it whenever you want without blocking.
	//
	// Note: can block if there is no space to buffer the message.
	SendWaitable(m amqp.Message) <-chan Outcome

	// SendForget buffers a message for sending and returns, with no notification of the outcome.
	//
	// Note: can block if there is no space to buffer the message.
	SendForget(m amqp.Message)

	// SendAsync puts a message in the send buffer and returns immediately. An
	// Outcome with Value = value will be sent to the ack channel when the remote
	// receiver has acknowledged the message or if there is an error.
	//
	// You can use the same ack channel for many calls to SendAsync(), possibly on
	// many Senders. The channel will receive the outcomes in the order they
	// become available. The channel should be buffered and/or served by dedicated
	// goroutines to avoid blocking the connection.
	//
	// If ack == nil no Outcome is sent.
	//
	// Note: can block if there is no space to buffer the message.
	SendAsync(m amqp.Message, ack chan<- Outcome, value interface{})

	SendAsyncTimeout(m amqp.Message, ack chan<- Outcome, value interface{}, timeout time.Duration)

	SendWaitableTimeout(m amqp.Message, timeout time.Duration) <-chan Outcome

	SendForgetTimeout(m amqp.Message, timeout time.Duration)

	SendSyncTimeout(m amqp.Message, timeout time.Duration) Outcome
	}

	// Outcome provides information about the outcome of sending a message.
	type Outcome struct {
	// Status of the message: was it sent, how was it acknowledged.
	Status SentStatus
	// Error is a local error if Status is Unsent or Unacknowledged, a remote error otherwise.
	Error error
	// Value provided by the application in SendAsync()
	Value interface{}
	}

	func (o Outcome) send(ack chan<- Outcome) {
	if ack != nil {
	ack <- o
	}
	}

	// SentStatus indicates the status of a sent message.
	type SentStatus int

	const (
	// Message was never sent
	Unsent SentStatus = iota
	// Message was sent but never acknowledged. It may or may not have been received.
	Unacknowledged
	// Message was accepted by the receiver (or was sent pre-settled, accept is assumed)
	Accepted
	// Message was rejected as invalid by the receiver
	Rejected
	// Message was not processed by the receiver but may be valid for a different receiver
	Released
	// Receiver responded with an unrecognized status.
	Unknown
	)

	// String human readable name for SentStatus.
	func (s SentStatus) String() string {
	switch s {
	case Unsent:
	return "unsent"
	case Unacknowledged:
	return "unacknowledged"
	case Accepted:
	return "accepted"
	case Rejected:
	return "rejected"
	case Released:
	return "released"
	case Unknown:
	return "unknown"
	default:
	return fmt.Sprintf("invalid(%d)", s)
	}
	}

	// Convert proton delivery state code to SentStatus value
	func sentStatus(d uint64) SentStatus {
	switch d {
	case proton.Accepted:
	return Accepted
	case proton.Rejected:
	return Rejected
	case proton.Released, proton.Modified:
	return Released
	default:
	return Unknown
	}
	}

	type sendable struct {
	m amqp.Message
	ack chan<- Outcome // Channel for acknowledgement of m
	v interface{} // Correlation value
	sent chan struct{} // Closed when m is encoded and will be sent
	}

	func (sm *sendable) unsent(err error) {
	Outcome{Unsent, err, sm.v}.send(sm.ack)
	}

	type sender struct {
	link
	sending []*sendable
	}

	func newSender(ls linkSettings) *sender {
	s := &sender{link: link{linkSettings: ls}}
	s.endpoint.init(s.link.pLink.String())
	s.handler().addLink(s.pLink, s)
	s.link.pLink.Open()
	return s
	}

	// Called in handler goroutine
	func (s sender) startSend(sm sendable) {
	s.sending = append(s.sending, sm)
	s.trySend()
	}

	// Called in handler goroutine
	func (s *sender) trySend() {
	for s.pLink.Credit() > 0 && len(s.sending) > 0 {
	sm := s.sending[0]
	s.sending = s.sending[1:]
	s.send(sm)
	}
	}

	// Called in handler goroutine with credit > 0
	func (s sender) send(sm sendable) {
	if err := s.Error(); err != nil {
	sm.unsent(err)
	return
	}
	bytes, err := s.session.connection.mc.Encode(sm.m, nil)
	close(sm.sent) // Safe to re-use sm.m now
	if err != nil {
	sm.unsent(err)
	return
	}
	d, err := s.pLink.SendMessageBytes(bytes)
	if err != nil {
	sm.unsent(err)
	return
	}
	if s.SndSettle() == SndSettled \|\| (s.SndSettle() == SndMixed && sm.ack == nil) {
	d.Settle() // Pre-settled
	Outcome{Accepted, nil, sm.v}.send(sm.ack) // Assume accepted
	} else {
	// Register with handler to receive the remote outcome
	s.handler().sent[d] = sm
	}
	}

	func (s sender) timeoutSend(sm sendable) {
	for i, sm2 := range s.sending {
	if sm2 == sm {
	n := copy(s.sending[i:], s.sending[i+1:])
	s.sending = s.sending[:i+n] // delete
	close(sm.sent)
	return
	}
	}
	}

	func (s *sender) SendAsyncTimeout(m amqp.Message, ack chan<- Outcome, v interface{}, t time.Duration) {
	sm := &sendable{m, ack, v, make(chan struct{})}
	s.engine().Inject(func() { s.startSend(sm) })
	select {
	case <-sm.sent: // OK
	case <-After(t): // Try to timeout sm
	s.engine().Inject(func() { s.timeoutSend(sm) })
	}
	}

	func (s *sender) SendWaitableTimeout(m amqp.Message, t time.Duration) <-chan Outcome {
	out := make(chan Outcome, 1)
	s.SendAsyncTimeout(m, out, nil, t)
	return out
	}

	func (s *sender) SendForgetTimeout(m amqp.Message, t time.Duration) {
	s.SendAsyncTimeout(m, nil, nil, t)
	}

	func (s *sender) SendSyncTimeout(m amqp.Message, t time.Duration) Outcome {
	deadline := time.Now().Add(t)
	ack := s.SendWaitableTimeout(m, t)
	t = deadline.Sub(time.Now()) // Adjust for time already spent.
	if t < 0 {
	t = 0
	}
	if out, err := timedReceive(ack, t); err == nil {
	return out.(Outcome)
	} else {
	if err == Closed && s.Error() != nil {
	err = s.Error()
	}
	return Outcome{Unacknowledged, err, nil}
	}
	}

	func (s *sender) SendAsync(m amqp.Message, ack chan<- Outcome, v interface{}) {
	s.SendAsyncTimeout(m, ack, v, Forever)
	}

	func (s *sender) SendWaitable(m amqp.Message) <-chan Outcome {
	return s.SendWaitableTimeout(m, Forever)
	}

	func (s *sender) SendForget(m amqp.Message) {
	s.SendForgetTimeout(m, Forever)
	}

	func (s *sender) SendSync(m amqp.Message) Outcome {
	return <-s.SendWaitable(m)
	}

	// handler goroutine
	func (s *sender) closed(err error) error {
	for _, sm := range s.sending {
	close(sm.sent)
	}
	s.sending = nil
	return s.link.closed(err)
	}

	// IncomingSender is sent on the Connection.Incoming() channel when there is
	// an incoming request to open a sender link.
	type IncomingSender struct {
	incoming
	linkSettings
	}

	func newIncomingSender(sn session, pLink proton.Link) IncomingSender {
	return &IncomingSender{
	incoming: makeIncoming(pLink),
	linkSettings: makeIncomingLinkSettings(pLink, sn),
	}
	}

	// Accept accepts an incoming sender endpoint
	func (in *IncomingSender) Accept() Endpoint {
	return in.accept(func() Endpoint { return newSender(in.linkSettings) })
	}

	// Call in injected functions to check if the sender is valid.
	func (s *sender) valid() bool {
	s2, ok := s.handler().links[s.pLink].(*sender)
	return ok && s2 == s
	}