Pfthb

I am fairly new to Go, I want to make a pipeline that translate every requests I receive by send it to first queue (TEST), and get the final result from the last queue (RESULT) and send it back as a response.

The problem I am facing is, the response never wait til all result back from the queue. Here is the code:

func main() 
 requests := int3, 4, 5, 6, 7
 var wg sync.WaitGroup
 wg.Add(1)
 resArr := string
 go func() 
 for _, r := range requests 
 rabbitSend("TEST", r)
 resArr = append(resArr, <-rabbitReceive("RESULT"))
 
 defer wg.Done()
 ()
 wg.Wait()

 log.Println("Result", resArr)

rabbitSend method:

func rabbitSend(queueName string, msg int) 
 conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
 failOnError(err, "Failed to connect to RabbitMQ")
 defer conn.Close()

 ch, err := conn.Channel()
 failOnError(err, "Failed to open a channel")
 defer ch.Close()

 q, err := ch.QueueDeclare(
 queueName, // name
 true, // durable
 false, // delete when unused
 false, // exclusive
 false, // no-wait
 nil, // arguments
 )
 failOnError(err, "Failed to declare a queue")

 body, _ := json.Marshal(msg)
 err = ch.Publish(
 "", // exchange
 q.Name, // routing key
 false, // mandatory
 false, // immediate
 amqp.Publishing
 ContentType: "application/json",
 Body: byte(body),
 )
 log.Printf("[x] Sent %s to %s", body, q.Name)
 failOnError(err, "Failed to publish a message")

rabbitReceive method:

func rabbitReceive(queueName string) <-chan string 
 conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
 failOnError(err, "Failed to connect to RabbitMQ")
 defer conn.Close()

 ch, err := conn.Channel()
 failOnError(err, "Failed to open a channel")
 defer ch.Close()

 q, err := ch.QueueDeclare(
 queueName, // name
 true, // durable
 false, // delete when usused
 false, // exclusive
 false, // no-waits
 nil, // arguments
 )
 failOnError(err, "Failed to declare a queue")

 msgs, err := ch.Consume(
 q.Name, // queue
 "", // consumer
 true, // auto-ack
 false, // exclusive
 false, // no-local
 false, // no-wait
 nil, // args
 )
 failOnError(err, "Failed to register a consumer")

 resCh := make(chan string)
 go func() 
 for d := range msgs 
 log.Printf("Received a message: %v from %v", string(d.Body), q.Name)
 resCh <- string(d.Body)
 
 close(resCh)
 ()
 return resCh

Here is what I get when I run the program:

2018/11/12 05:11:54 [x] Sent 3 to TEST
2018/11/12 05:11:54 [x] Sent 4 to TEST
2018/11/12 05:11:54 Received a message: 9 from RESULT
2018/11/12 05:11:54 [x] Sent 5 to TEST
2018/11/12 05:11:54 [x] Sent 6 to TEST
2018/11/12 05:11:54 Received a message: 15 from RESULT
2018/11/12 05:11:54 [x] Sent 7 to TEST
2018/11/12 05:11:54 Received a message: 18 from RESULT
2018/11/12 05:11:54 Result [ 9 15 18]

What I want is that, I receive the result exactly after I send the request, so the request will not get the wrong result as a response. Something like:

2018/11/12 05:11:54 [x] Sent 3 to TEST
2018/11/12 05:11:54 Received a message: 9 from RESULT
2018/11/12 05:11:54 [x] Sent 4 to TEST
2018/11/12 05:11:54 Received a message: 12 from RESULT
2018/11/12 05:11:54 [x] Sent 5 to TEST
2018/11/12 05:11:54 Received a message: 15 from RESULT
2018/11/12 05:11:54 [x] Sent 6 to TEST
2018/11/12 05:11:54 Received a message: 18 from RESULT
2018/11/12 05:11:54 [x] Sent 7 to TEST
2018/11/12 05:11:54 Received a message: 21 from RESULT
2018/11/12 05:11:54 Result [ 9 12 15 18 21]

I believe I did not use goroutine or sync.WaitGroup correctly here. Thanks in advance :)

Modify your func rabbitReceive(queueName string) <-chan string as below:

 func rabbitReceive(queueName string) <-chan string 
 conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
 failOnError(err, "Failed to connect to RabbitMQ")

 ch, err := conn.Channel()
 failOnError(err, "Failed to open a channel")

 q, err := ch.QueueDeclare(
 queueName, // name
 true, // durable
 false, // delete when usused
 false, // exclusive
 false, // no-waits
 nil, // arguments
 )
 failOnError(err, "Failed to declare a queue")

 msgs, err := ch.Consume(
 q.Name, // queue
 "", // consumer
 true, // auto-ack
 false, // exclusive
 false, // no-local
 false, // no-wait
 nil, // args
 )
 failOnError(err, "Failed to register a consumer")

 resCh := make(chan string)
 go func() 
 d := <-msgs
 log.Printf("Received a message: %v from %v", string(d.Body), q.Name)
 resCh <- string(d.Body)
 conn.Close()
 ch.Close()
 close(resCh)
 ()
 return resCh

The reason previous code cause you problem was defer ch.Close(). ch closes before response was written to resCh.

following up on @nightfury1204 great answer, you are indeed closing ch before writing to resCh. just one thing, in the go routine you want to go over all the messages so a better way to do it will be:

go func() 
 for d := range msgs 
 log.Printf("Received a message: %v from %v", string(d.Body), q.Name)
 resCh <- string(d.Body) 
 
 conn.Close()
 ch.Close()
 close(resCh)
 ()