Documentation
¶
Overview ¶
Package amqp is the AMQP implementation of an event stream.
Index ¶
Examples ¶
Constants ¶
View Source
const ( // ExchangeNameOption is the exchange name option. ExchangeNameOption = "exchange.name" // ExchangeTypeOption is the exchange type option. ExchangeTypeOption = "exchange.type" // ExchangeDurable is the exchange durable option. ExchangeDurable = "exchange.durable" // ExchangeAutoDelete is the exchange auto delete option. ExchangeAutoDelete = "exchange.auto_delete" // ExchangeInternal is the exchange internal option. ExchangeInternal = "exchange.internal" // ExchangeNoWait is the exchange no wait option. ExchangeNoWait = "exchange.no_wait" // RoutingKeyOption is the routing key option. RoutingKeyOption = "routing_key" )
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type AMQP ¶
type AMQP struct {
// contains filtered or unexported fields
}
AMQP is the wrapper for the AMQP library.
func NewAMQP ¶
func NewAMQP(conn *amqp.Connection) *AMQP
NewAMQP creates a new instance of AMQP.
Example ¶
package main
import (
"context"
"fmt"
"os"
"github.com/artsv79/outboxer"
amqpOut "github.com/artsv79/outboxer/es/amqp"
amqp "github.com/rabbitmq/amqp091-go"
)
func main() {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
conn, err := amqp.Dial(os.Getenv("ES_DSN"))
if err != nil {
fmt.Printf("failed to connect to amqp: %s", err)
return
}
defer conn.Close()
es := amqpOut.NewAMQP(conn)
// this is done internally by outboxer
if err := es.Send(ctx, &outboxer.OutboxMessage{
Payload: []byte("test payload"),
Options: map[string]interface{}{
amqpOut.ExchangeNameOption: "test",
amqpOut.ExchangeTypeOption: "topic",
amqpOut.RoutingKeyOption: "test.send",
},
}); err != nil {
fmt.Printf("an error was not expected: %s", err)
return
}
}
Output:
Source Files
Click to show internal directories.
Click to hide internal directories.