Restructure go-micro layout and plugins

transport/rabbitmq/channel.go

@@ -0,0 +1,127 @@
+package rabbitmq
+
+//
+// All credit to Mondo
+//
+
+import (
+	"errors"
+
+	"github.com/nu7hatch/gouuid"
+	"github.com/streadway/amqp"
+)
+
+type rabbitMQChannel struct {
+	uuid       string
+	connection *amqp.Connection
+	channel    *amqp.Channel
+}
+
+func newRabbitChannel(conn *amqp.Connection) (*rabbitMQChannel, error) {
+	id, err := uuid.NewV4()
+	if err != nil {
+		return nil, err
+	}
+	rabbitCh := &rabbitMQChannel{
+		uuid:       id.String(),
+		connection: conn,
+	}
+	if err := rabbitCh.Connect(); err != nil {
+		return nil, err
+	}
+	return rabbitCh, nil
+
+}
+
+func (r *rabbitMQChannel) Connect() error {
+	var err error
+	r.channel, err = r.connection.Channel()
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (r *rabbitMQChannel) Close() error {
+	if r.channel == nil {
+		return errors.New("Channel is nil")
+	}
+	return r.channel.Close()
+}
+
+func (r *rabbitMQChannel) Publish(exchange, key string, message amqp.Publishing) error {
+	if r.channel == nil {
+		return errors.New("Channel is nil")
+	}
+	return r.channel.Publish(exchange, key, false, false, message)
+}
+
+func (r *rabbitMQChannel) DeclareExchange(exchange string) error {
+	return r.channel.ExchangeDeclare(
+		exchange, // name
+		"topic",  // kind
+		false,    // durable
+		false,    // autoDelete
+		false,    // internal
+		false,    // noWait
+		nil,      // args
+	)
+}
+
+func (r *rabbitMQChannel) DeclareQueue(queue string) error {
+	_, err := r.channel.QueueDeclare(
+		queue, // name
+		false, // durable
+		true,  // autoDelete
+		false, // exclusive
+		false, // noWait
+		nil,   // args
+	)
+	return err
+}
+
+func (r *rabbitMQChannel) DeclareDurableQueue(queue string) error {
+	_, err := r.channel.QueueDeclare(
+		queue, // name
+		true,  // durable
+		false, // autoDelete
+		false, // exclusive
+		false, // noWait
+		nil,   // args
+	)
+	return err
+}
+
+func (r *rabbitMQChannel) DeclareReplyQueue(queue string) error {
+	_, err := r.channel.QueueDeclare(
+		queue, // name
+		false, // durable
+		true,  // autoDelete
+		true,  // exclusive
+		false, // noWait
+		nil,   // args
+	)
+	return err
+}
+
+func (r *rabbitMQChannel) ConsumeQueue(queue string) (<-chan amqp.Delivery, error) {
+	return r.channel.Consume(
+		queue,  // queue
+		r.uuid, // consumer
+		true,   // autoAck
+		false,  // exclusive
+		false,  // nolocal
+		false,  // nowait
+		nil,    // args
+	)
+}
+
+func (r *rabbitMQChannel) BindQueue(queue, exchange string) error {
+	return r.channel.QueueBind(
+		queue,    // name
+		queue,    // key
+		exchange, // exchange
+		false,    // noWait
+		nil,      // args
+	)
+}

transport/rabbitmq/connection.go

@@ -0,0 +1,147 @@
+package rabbitmq
+
+//
+// All credit to Mondo
+//
+
+import (
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/streadway/amqp"
+)
+
+var (
+	DefaultExchange  = "micro"
+	DefaultRabbitURL = "amqp://guest:guest@127.0.0.1:5672"
+)
+
+type rabbitMQConn struct {
+	Connection      *amqp.Connection
+	Channel         *rabbitMQChannel
+	ExchangeChannel *rabbitMQChannel
+	notify          chan bool
+	exchange        string
+	url             string
+
+	connected bool
+
+	mtx    sync.Mutex
+	close  chan bool
+	closed bool
+}
+
+func newRabbitMQConn(exchange string, urls []string) *rabbitMQConn {
+	var url string
+
+	if len(urls) > 0 && strings.HasPrefix(urls[0], "amqp://") {
+		url = urls[0]
+	} else {
+		url = DefaultRabbitURL
+	}
+
+	if len(exchange) == 0 {
+		exchange = DefaultExchange
+	}
+
+	return &rabbitMQConn{
+		exchange: exchange,
+		url:      url,
+		notify:   make(chan bool, 1),
+		close:    make(chan bool),
+	}
+}
+
+func (r *rabbitMQConn) Init() chan bool {
+	go r.Connect(r.notify)
+	return r.notify
+}
+
+func (r *rabbitMQConn) Connect(connected chan bool) {
+	for {
+		if err := r.tryToConnect(); err != nil {
+			time.Sleep(1 * time.Second)
+			continue
+		}
+		connected <- true
+		r.connected = true
+		notifyClose := make(chan *amqp.Error)
+		r.Connection.NotifyClose(notifyClose)
+
+		// Block until we get disconnected, or shut down
+		select {
+		case <-notifyClose:
+			// Spin around and reconnect
+			r.connected = false
+		case <-r.close:
+			// Shut down connection
+			if err := r.Connection.Close(); err != nil {
+			}
+			r.connected = false
+			return
+		}
+	}
+}
+
+func (r *rabbitMQConn) IsConnected() bool {
+	return r.connected
+}
+
+func (r *rabbitMQConn) Close() {
+	r.mtx.Lock()
+	defer r.mtx.Unlock()
+
+	if r.closed {
+		return
+	}
+
+	close(r.close)
+	r.closed = true
+}
+
+func (r *rabbitMQConn) tryToConnect() error {
+	var err error
+	r.Connection, err = amqp.Dial(r.url)
+	if err != nil {
+		return err
+	}
+	r.Channel, err = newRabbitChannel(r.Connection)
+	if err != nil {
+		return err
+	}
+	r.Channel.DeclareExchange(r.exchange)
+	r.ExchangeChannel, err = newRabbitChannel(r.Connection)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (r *rabbitMQConn) Consume(queue string) (<-chan amqp.Delivery, error) {
+	consumerChannel, err := newRabbitChannel(r.Connection)
+	if err != nil {
+		return nil, err
+	}
+
+	err = consumerChannel.DeclareQueue(queue)
+	if err != nil {
+		return nil, err
+	}
+
+	deliveries, err := consumerChannel.ConsumeQueue(queue)
+	if err != nil {
+		return nil, err
+	}
+
+	err = consumerChannel.BindQueue(queue, r.exchange)
+	if err != nil {
+		return nil, err
+	}
+
+	return deliveries, nil
+}
+
+func (r *rabbitMQConn) Publish(exchange, key string, msg amqp.Publishing) error {
+	return r.ExchangeChannel.Publish(exchange, key, msg)
+}

transport/rabbitmq/rabbitmq.go

@@ -0,0 +1,239 @@
+package rabbitmq
+
+import (
+	"fmt"
+	"sync"
+	"time"
+
+	"errors"
+	uuid "github.com/nu7hatch/gouuid"
+	"github.com/streadway/amqp"
+
+	"github.com/myodc/go-micro/transport"
+)
+
+type rmqtport struct {
+	conn  *rabbitMQConn
+	addrs []string
+}
+
+type rmqtportClient struct {
+	once    sync.Once
+	rt      *rmqtport
+	addr    string
+	replyTo string
+
+	sync.Mutex
+	inflight map[string]chan amqp.Delivery
+}
+
+type rmqtportSocket struct {
+	conn *rabbitMQConn
+	d    *amqp.Delivery
+}
+
+type rmqtportListener struct {
+	conn *rabbitMQConn
+	addr string
+}
+
+func (r *rmqtportClient) init() {
+	<-r.rt.conn.Init()
+	if err := r.rt.conn.Channel.DeclareReplyQueue(r.replyTo); err != nil {
+		return
+	}
+	deliveries, err := r.rt.conn.Channel.ConsumeQueue(r.replyTo)
+	if err != nil {
+		return
+	}
+	go func() {
+		for delivery := range deliveries {
+			go r.handle(delivery)
+		}
+	}()
+}
+
+func (r *rmqtportClient) handle(delivery amqp.Delivery) {
+	ch := r.getReq(delivery.CorrelationId)
+	if ch == nil {
+		return
+	}
+	select {
+	case ch <- delivery:
+	default:
+	}
+}
+
+func (r *rmqtportClient) putReq(id string) chan amqp.Delivery {
+	r.Lock()
+	ch := make(chan amqp.Delivery, 1)
+	r.inflight[id] = ch
+	r.Unlock()
+	return ch
+}
+
+func (r *rmqtportClient) getReq(id string) chan amqp.Delivery {
+	r.Lock()
+	defer r.Unlock()
+	if ch, ok := r.inflight[id]; ok {
+		delete(r.inflight, id)
+		return ch
+	}
+	return nil
+}
+
+func (r *rmqtportClient) Send(m *transport.Message) (*transport.Message, error) {
+	r.once.Do(r.init)
+
+	if !r.rt.conn.IsConnected() {
+		return nil, errors.New("Not connected to AMQP")
+	}
+
+	id, err := uuid.NewV4()
+	if err != nil {
+		return nil, err
+	}
+
+	replyChan := r.putReq(id.String())
+
+	headers := amqp.Table{}
+
+	for k, v := range m.Header {
+		headers[k] = v
+	}
+
+	message := amqp.Publishing{
+		CorrelationId: id.String(),
+		Timestamp:     time.Now().UTC(),
+		Body:          m.Body,
+		ReplyTo:       r.replyTo,
+		Headers:       headers,
+	}
+
+	if err := r.rt.conn.Publish("micro", r.addr, message); err != nil {
+		r.getReq(id.String())
+		return nil, err
+	}
+
+	select {
+	case d := <-replyChan:
+		mr := &transport.Message{
+			Header: make(map[string]string),
+			Body:   d.Body,
+		}
+
+		for k, v := range d.Headers {
+			mr.Header[k] = fmt.Sprintf("%v", v)
+		}
+
+		return mr, nil
+	case <-time.After(time.Second * 10):
+		return nil, errors.New("timed out")
+	}
+}
+
+func (r *rmqtportClient) Close() error {
+	return nil
+}
+
+func (r *rmqtportSocket) Recv(m *transport.Message) error {
+	if m == nil {
+		return errors.New("message passed in is nil")
+	}
+
+	mr := &transport.Message{
+		Header: make(map[string]string),
+		Body:   r.d.Body,
+	}
+
+	for k, v := range r.d.Headers {
+		mr.Header[k] = fmt.Sprintf("%v", v)
+	}
+
+	*m = *mr
+	return nil
+}
+
+func (r *rmqtportSocket) Send(m *transport.Message) error {
+	msg := amqp.Publishing{
+		CorrelationId: r.d.CorrelationId,
+		Timestamp:     time.Now().UTC(),
+		Body:          m.Body,
+		Headers:       amqp.Table{},
+	}
+
+	for k, v := range m.Header {
+		msg.Headers[k] = v
+	}
+
+	return r.conn.Publish("", r.d.ReplyTo, msg)
+}
+
+func (r *rmqtportSocket) Close() error {
+	return nil
+}
+
+func (r *rmqtportListener) Addr() string {
+	return r.addr
+}
+
+func (r *rmqtportListener) Close() error {
+	r.conn.Close()
+	return nil
+}
+
+func (r *rmqtportListener) Accept(fn func(transport.Socket)) error {
+	deliveries, err := r.conn.Consume(r.addr)
+	if err != nil {
+		return err
+	}
+
+	handler := func(d amqp.Delivery) {
+		fn(&rmqtportSocket{
+			d:    &d,
+			conn: r.conn,
+		})
+	}
+
+	for d := range deliveries {
+		go handler(d)
+	}
+
+	return nil
+}
+
+func (r *rmqtport) Dial(addr string) (transport.Client, error) {
+	id, err := uuid.NewV4()
+	if err != nil {
+		return nil, err
+	}
+
+	return &rmqtportClient{
+		rt:       r,
+		addr:     addr,
+		inflight: make(map[string]chan amqp.Delivery),
+		replyTo:  fmt.Sprintf("replyTo-%s", id.String()),
+	}, nil
+}
+
+func (r *rmqtport) Listen(addr string) (transport.Listener, error) {
+	id, err := uuid.NewV4()
+	if err != nil {
+		return nil, err
+	}
+
+	conn := newRabbitMQConn("", r.addrs)
+	<-conn.Init()
+
+	return &rmqtportListener{
+		addr: id.String(),
+		conn: conn,
+	}, nil
+}
+
+func NewTransport(addrs []string, opt ...transport.Option) transport.Transport {
+	return &rmqtport{
+		conn:  newRabbitMQConn("", addrs),
+		addrs: addrs,
+	}
+}