Vfrdtyky

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 := int{3, 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 before :)

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)

    }()