func (t *MultiTransport) Done() chan struct{} {
	return t.stopchan
}
Done gets a channel which is closed when the
transport has successfully stopped.
package sqs
import (
	"fmt"
	"sync"
	"time"
	"github.com/matryer/vice"
	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/sqs"
	"github.com/aws/aws-sdk-go/service/sqs/sqsiface"
)
type MultiTransport struct {
	writers       int
	batchSize     int
	batchInterval time.Duration
	wg sync.WaitGroup
	sm        sync.Mutex
	sendChans map[string]chan []byte
	rm           sync.Mutex
	receiveChans map[string]chan []byte
	errChan     chan error
	stopchan    chan struct{}
	stopPubChan chan struct{}
	stopSubChan chan struct{}
	NewService func(region string) (sqsiface.SQSAPI, error)
}
var _ vice.Transport = (*MultiTransport)(nil)
func NewMulti(writers, batchSize int, batchInterval time.Duration) *MultiTransport {
	const defaultWriter = 2
	if writers == 0 {
		writers = defaultWriter
	}
	const maxBatchSize = 10
	if batchSize > 10 {
		batchSize = maxBatchSize
	}
	const defaultInterval = 200 * time.Millisecond
	if batchInterval == 0 {
		batchInterval = defaultInterval
	}
	return &MultiTransport{
		writers:       writers,
		batchSize:     batchSize,
		batchInterval: batchInterval,
		sendChans:     make(map[string]chan []byte),
		receiveChans:  make(map[string]chan []byte),
		errChan:       make(chan error, 10),
		stopchan:      make(chan struct{}),
		stopPubChan:   make(chan struct{}),
		stopSubChan:   make(chan struct{}),
		NewService: func(region string) (sqsiface.SQSAPI, error) {
			awsConfig := aws.NewConfig().WithRegion(region)
			s, err := session.NewSession(awsConfig)
			if err != nil {
				return nil, err
			}
			return sqs.New(s), nil
		},
	}
}
func (t *MultiTransport) Receive(name string) <-chan []byte {
	t.rm.Lock()
	defer t.rm.Unlock()
	ch, ok := t.receiveChans[name]
	if ok {
		return ch
	}
	ch, err := t.makeSubscriber(name)
	if err != nil {
		t.errChan <- &vice.Err{Name: name, Err: err}
		return make(chan []byte)
	}
	t.receiveChans[name] = ch
	return ch
}
func (t *MultiTransport) makeSubscriber(name string) (chan []byte, error) {
	region := RegionFromURL(name)
	svc, err := t.NewService(region)
	if err != nil {
		return nil, err
	}
	ch := make(chan []byte, 1024)
	params := &sqs.ReceiveMessageInput{
		QueueUrl:            aws.String(name),
		MaxNumberOfMessages: aws.Int64(1),
		VisibilityTimeout:   aws.Int64(1),
		WaitTimeSeconds:     aws.Int64(1),
	}
	go func() {
		for {
			select {
			case <-t.stopSubChan:
				return
			default:
				resp, err := svc.ReceiveMessage(params)
				if err != nil {
					t.errChan <- &vice.Err{Name: name, Err: err}
					continue
				}
				if len(resp.Messages) > 0 {
					for _, m := range resp.Messages {
						if m.ReceiptHandle != nil {
							delParams := &sqs.DeleteMessageInput{
								QueueUrl:      aws.String(name),
								ReceiptHandle: aws.String(*m.ReceiptHandle),
							}
							_, err := svc.DeleteMessage(delParams)
							if err != nil {
								t.errChan <- &vice.Err{Name: name, Err: err}
								continue
							}
						}
						ch <- []byte(*m.Body)
					}
				}
			}
		}
	}()
	return ch, nil
}
func (t *MultiTransport) Send(name string) chan<- []byte {
	t.sm.Lock()
	defer t.sm.Unlock()
	ch, ok := t.sendChans[name]
	if ok {
		return ch
	}
	ch, err := t.makePublishers(name)
	if err != nil {
		t.errChan <- &vice.Err{Name: name, Err: err}
		return make(chan []byte)
	}
	t.sendChans[name] = ch
	return ch
}
func (t *MultiTransport) makePublishers(name string) (chan []byte, error) {
	region := RegionFromURL(name)
	queue := make(chan writerMsg)
	err := newWriterGroup(t.writers, t.stopchan, t.errChan, queue, t.NewService, region)
	if err != nil {
		return nil, err
	}
	ch := make(chan []byte, 1024)
	t.wg.Add(1)
	go func() {
		defer t.wg.Done()
		var accum sendMsgBatch
		defer func() {
			if len(accum) == 0 {
				return
			}
			queue <- writerMsg{batch: accum, name: name}
			time.Sleep(50 * time.Millisecond)
		}()
		for {
			select {
			case <-t.stopPubChan:
				if len(ch) != 0 {
					continue
				}
				return
			case msg := <-ch:
				id := fmt.Sprintf("%d", time.Now().UnixNano())
				accum = append(accum, sqs.SendMessageBatchRequestEntry{
					MessageBody: aws.String(string(msg)),
					Id:          aws.String(id),
				})
				if t.batchSize == 0 {
					tmp := make(sendMsgBatch, len(accum))
					copy(tmp, accum)
					queue <- writerMsg{batch: tmp, name: name}
					accum = make(sendMsgBatch, 0, t.batchSize)
					continue
				}
				if len(accum) < t.batchSize {
					continue
				}
				tmp := make(sendMsgBatch, len(accum))
				copy(tmp, accum)
				queue <- writerMsg{batch: tmp, name: name}
				accum = make(sendMsgBatch, 0, t.batchSize)
			case <-time.After(t.batchInterval):
				if t.batchSize == 0 || len(accum) == 0 {
					continue
				}
				tmp := make(sendMsgBatch, len(accum))
				copy(tmp, accum)
				queue <- writerMsg{batch: tmp, name: name}
				accum = make(sendMsgBatch, 0, t.batchSize)
			}
		}
	}()
	return ch, nil
}
func (t *MultiTransport) ErrChan() <-chan error {
	return t.errChan
}
func (t *MultiTransport) Stop() {
	close(t.stopSubChan)
	close(t.stopPubChan)
	t.wg.Wait()
	close(t.stopchan)
}