89 lines
1.7 KiB
Go
89 lines
1.7 KiB
Go
package amqp
|
|
|
|
import (
|
|
"bytes"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log"
|
|
|
|
"github.com/streadway/amqp"
|
|
)
|
|
|
|
const (
|
|
EVENT_BASKET_CHECKOUT = "event.BasketCheckoutEvent"
|
|
)
|
|
|
|
type Message map[string]interface{}
|
|
|
|
func Serialize(msg any /*Message*/) ([]byte, error) {
|
|
var b bytes.Buffer
|
|
encoder := json.NewEncoder(&b)
|
|
err := encoder.Encode(msg)
|
|
return b.Bytes(), err
|
|
}
|
|
|
|
func Deserialize(b []byte) (Message, error) {
|
|
var msg Message
|
|
buf := bytes.NewBuffer(b)
|
|
decoder := json.NewDecoder(buf)
|
|
err := decoder.Decode(&msg)
|
|
return msg, err
|
|
}
|
|
|
|
func NewExchange(chn *amqp.Channel, name string) error {
|
|
err := chn.ExchangeDeclare(
|
|
name,
|
|
"direct", // type
|
|
true, // durable
|
|
false, // auto-deleted
|
|
false, // internal
|
|
false, // no-wait
|
|
nil, // arguments
|
|
) // FIXME extend arguments when needed...
|
|
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func Publish(chn *amqp.Channel, name, routingKey string, msg any) error {
|
|
// var jsonData []byte
|
|
jsonData, err := Serialize(msg)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
msgBody := fmt.Sprintf(`{"event":"%T","data":%s}`, msg, jsonData) // FIXME %T - simplify
|
|
chn.Publish(
|
|
name, // exchange name
|
|
routingKey, // routing key
|
|
false, // mandatory
|
|
false, // immediate
|
|
amqp.Publishing{
|
|
ContentType: "application/json",
|
|
Body: []byte(msgBody),
|
|
},
|
|
)
|
|
|
|
return nil
|
|
}
|
|
|
|
func BindQueueToExchange(chn *amqp.Channel, queueName, exchName, routingKey string) error {
|
|
err := chn.QueueBind(
|
|
queueName, // queue name
|
|
routingKey, // routing key
|
|
exchName, // exchange name FIXME: use const: ebEventsExchange
|
|
false,
|
|
nil,
|
|
)
|
|
if err != nil {
|
|
log.Printf("Failed to bind a queue: %s\n", queueName)
|
|
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|