go/microservices
1
0
Fork 0
mirror of https://github.com/ebosas/microservices.git synced 2025-08-24 20:08:55 +02:00
Code Issues Releases Activity

Comments

Browse Source
This commit is contained in:
ebosas
2021-06-08 21:41:51 +03:00
parent 0bab170f33
commit de11f113d7
3 changed files with 8 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
cmd/backend/backend.go
View File

@@ -32,14 +32,14 @@ func main() {
log.Fatalf("declare exchange: %s", err) log.Fatalf("declare exchange: %s", err)
} }
// Start a Rabbit consumer with a message processing handler. // Start a Rabbit consumer with a handler for printing messages.
conn.StartConsumer(conf.Exchange, conf.QueueBack, conf.KeyBack, receiveMessages) conn.StartConsumer(conf.Exchange, conf.QueueBack, conf.KeyBack, printMessages)
publishInput(conn) publishInput(conn)
} }
// receiveMessages prints messages to stdout. // printMessages prints messages to stdout.
func receiveMessages(d amqp.Delivery) bool { func printMessages(d amqp.Delivery) bool {
var message models.Message var message models.Message
err := json.Unmarshal(d.Body, &message) err := json.Unmarshal(d.Body, &message)
if err != nil { if err != nil {
@@ -55,7 +55,6 @@ func receiveMessages(d amqp.Delivery) bool {
// a Rabbit exchange with the front-end and database routing keys. // a Rabbit exchange with the front-end and database routing keys.
func publishInput(conn *rabbit.Conn) { func publishInput(conn *rabbit.Conn) {
reader := bufio.NewReader(os.Stdin) reader := bufio.NewReader(os.Stdin)
for { for {
input, _ := reader.ReadString('\n') input, _ := reader.ReadString('\n')
input = strings.TrimSpace(input) input = strings.TrimSpace(input)
@@ -68,6 +67,7 @@ func publishInput(conn *rabbit.Conn) {
if err != nil { if err != nil {
log.Fatalf("marshal message: %s", err) log.Fatalf("marshal message: %s", err)
} }
err = conn.Publish(conf.Exchange, conf.KeyFront+"."+conf.KeyDB, message) err = conn.Publish(conf.Exchange, conf.KeyFront+"."+conf.KeyDB, message)
if err != nil { if err != nil {
log.Fatalf("publish message: %s", err) log.Fatalf("publish message: %s", err)

1
internal/rabbit/consume.go
View File

@@ -13,6 +13,7 @@ import (
// The queue is created (or connected to, if exists) and bound to an exchange. // The queue is created (or connected to, if exists) and bound to an exchange.
// Used for durable queues in the main go routine. // Used for durable queues in the main go routine.
func (conn *Conn) StartConsumer(exch, qName, rKey string, handler func(amqp.Delivery) bool) error { func (conn *Conn) StartConsumer(exch, qName, rKey string, handler func(amqp.Delivery) bool) error {
// Declare a durable queue
_, err := conn.Channel.QueueDeclare(qName, true, false, false, false, nil) _, err := conn.Channel.QueueDeclare(qName, true, false, false, false, nil)
if err != nil { if err != nil {
return fmt.Errorf("queue declare: %v", err) return fmt.Errorf("queue declare: %v", err)

4
internal/rabbit/publish.go
View File

@@ -6,8 +6,8 @@ import (
"github.com/streadway/amqp" "github.com/streadway/amqp"
) )
// Publish publishes a message to a Rabbit exchange using a channel // Publish publishes a message to a Rabbit exchange using the main channel.
// opened upon connection to Rabbit. For use in the main go routine. // For use in the main go routine.
func (conn Conn) Publish(exch, rKey string, message []byte) error { func (conn Conn) Publish(exch, rKey string, message []byte) error {
return PublishInChannel(conn.Channel, exch, rKey, message) return PublishInChannel(conn.Channel, exch, rKey, message)
} }