tile38/internal/endpoint/amqp.go

138 lines
2.5 KiB
Go
Raw Normal View History

2017-03-15 20:45:35 +03:00
package endpoint
import (
2018-04-19 19:25:39 +03:00
"fmt"
"net"
2017-03-15 20:45:35 +03:00
"sync"
"time"
"github.com/streadway/amqp"
)
const amqpExpiresAfter = time.Second * 30
2017-03-15 20:45:35 +03:00
2018-04-19 19:25:39 +03:00
// AMQPConn is an endpoint connection
type AMQPConn struct {
2017-03-15 20:45:35 +03:00
mu sync.Mutex
ep Endpoint
conn *amqp.Connection
channel *amqp.Channel
ex bool
t time.Time
}
2018-04-19 19:25:39 +03:00
// Expired returns true if the connection has expired
func (conn *AMQPConn) Expired() bool {
2017-03-15 20:45:35 +03:00
conn.mu.Lock()
defer conn.mu.Unlock()
if !conn.ex {
2018-04-19 19:25:39 +03:00
if time.Now().Sub(conn.t) > amqpExpiresAfter {
2017-03-15 20:45:35 +03:00
conn.ex = true
conn.close()
}
}
return conn.ex
}
2018-04-19 19:25:39 +03:00
func (conn *AMQPConn) close() {
2017-03-15 20:45:35 +03:00
if conn.conn != nil {
conn.conn.Close()
conn.conn = nil
conn.channel = nil
}
}
2018-04-19 19:25:39 +03:00
// Send sends a message
func (conn *AMQPConn) Send(msg string) error {
2017-03-15 20:45:35 +03:00
conn.mu.Lock()
defer conn.mu.Unlock()
if conn.ex {
return errExpired
}
conn.t = time.Now()
if conn.conn == nil {
prefix := "amqp://"
if conn.ep.AMQP.SSL {
prefix = "amqps://"
}
var cfg amqp.Config
cfg.Dial = func(network, addr string) (net.Conn, error) {
return net.DialTimeout(network, addr, time.Second)
}
c, err := amqp.DialConfig(fmt.Sprintf("%s%s", prefix, conn.ep.AMQP.URI), cfg)
2017-03-15 20:45:35 +03:00
if err != nil {
return err
}
channel, err := c.Channel()
if err != nil {
return err
}
// Declare new exchange
if err := channel.ExchangeDeclare(
conn.ep.AMQP.QueueName,
conn.ep.AMQP.Type,
conn.ep.AMQP.Durable,
conn.ep.AMQP.AutoDelete,
conn.ep.AMQP.Internal,
conn.ep.AMQP.NoWait,
2017-03-15 20:45:35 +03:00
nil,
); err != nil {
return err
}
// Create queue if queue don't exists
if _, err := channel.QueueDeclare(
conn.ep.AMQP.QueueName,
conn.ep.AMQP.Durable,
conn.ep.AMQP.AutoDelete,
2017-03-15 20:45:35 +03:00
false,
conn.ep.AMQP.NoWait,
2017-03-15 20:45:35 +03:00
nil,
); err != nil {
return err
}
// Binding exchange to queue
if err := channel.QueueBind(
conn.ep.AMQP.QueueName,
conn.ep.AMQP.RouteKey,
conn.ep.AMQP.QueueName,
conn.ep.AMQP.NoWait,
2017-03-15 20:45:35 +03:00
nil,
); err != nil {
return err
}
conn.conn = c
conn.channel = channel
}
2018-04-19 19:25:39 +03:00
return conn.channel.Publish(
2017-03-15 20:45:35 +03:00
conn.ep.AMQP.QueueName,
conn.ep.AMQP.RouteKey,
conn.ep.AMQP.Mandatory,
conn.ep.AMQP.Immediate,
2017-03-15 20:45:35 +03:00
amqp.Publishing{
Headers: amqp.Table{},
ContentType: "application/json",
ContentEncoding: "",
Body: []byte(msg),
DeliveryMode: conn.ep.AMQP.DeliveryMode,
Priority: conn.ep.AMQP.Priority,
2017-03-15 20:45:35 +03:00
},
2018-04-19 19:25:39 +03:00
)
2017-03-15 20:45:35 +03:00
}
2018-04-19 19:25:39 +03:00
func newAMQPConn(ep Endpoint) *AMQPConn {
return &AMQPConn{
2017-03-15 20:45:35 +03:00
ep: ep,
t: time.Now(),
}
}