go/go-micro
1
0
Fork 0
mirror of https://github.com/go-micro/go-micro.git synced 2024-12-24 10:07:04 +02:00
Code Issues Releases Activity

Use direct reply queue

Browse Source
This commit is contained in:
Asim 2015-11-08 23:16:14 +00:00
parent 55ef1d4903
commit 3cf90d7825
2 changed files with 11 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
transport/rabbitmq/connection.go
View File

@ -20,7 +20,6 @@ var (
type rabbitMQConn struct { type rabbitMQConn struct {
Connection *amqp.Connection Connection *amqp.Connection
Channel *rabbitMQChannel Channel *rabbitMQChannel
ExchangeChannel *rabbitMQChannel
notify chan bool notify chan bool
exchange string exchange string
url string url string
@ -111,10 +110,6 @@ func (r *rabbitMQConn) tryToConnect() error {
return err return err
} }
r.Channel.DeclareExchange(r.exchange) r.Channel.DeclareExchange(r.exchange)
r.ExchangeChannel, err = newRabbitChannel(r.Connection)
if err != nil {
return err
}
return nil return nil
} }
@ -143,5 +138,5 @@ func (r *rabbitMQConn) Consume(queue string) (<-chan amqp.Delivery, error) {
} }
func (r *rabbitMQConn) Publish(exchange, key string, msg amqp.Publishing) error { func (r *rabbitMQConn) Publish(exchange, key string, msg amqp.Publishing) error {
return r.ExchangeChannel.Publish(exchange, key, msg) return r.Channel.Publish(exchange, key, msg)
} }

8
transport/rabbitmq/rabbitmq.go
View File

@ -12,6 +12,10 @@ import (
"github.com/myodc/go-micro/transport" "github.com/myodc/go-micro/transport"
) )
const (
directReplyQueue = "amq.rabbitmq.reply-to"
)
type rmqtport struct { type rmqtport struct {
conn *rabbitMQConn conn *rabbitMQConn
addrs []string addrs []string
@ -236,12 +240,10 @@ func (r *rmqtport) Listen(addr string) (transport.Listener, error) {
} }
func NewTransport(addrs []string, opt ...transport.Option) transport.Transport { func NewTransport(addrs []string, opt ...transport.Option) transport.Transport {
id, _ := uuid.NewV4()
return &rmqtport{ return &rmqtport{
conn: newRabbitMQConn("", addrs), conn: newRabbitMQConn("", addrs),
addrs: addrs, addrs: addrs,
replyTo: id.String(), replyTo: directReplyQueue,
inflight: make(map[string]chan amqp.Delivery), inflight: make(map[string]chan amqp.Delivery),
} }
} }