MultipleRequest sends a request message to multiple consumers
(topic string, req *proxy.Request, expectedResp int)
| 477 | |
| 478 | // MultipleRequest sends a request message to multiple consumers |
| 479 | func (r *RabbitmqBus) MultipleRequest(topic string, req *proxy.Request, expectedResp int) ([]*proxy.Response, error) { |
| 480 | |
| 481 | responses := make([]*proxy.Response, 0, expectedResp) |
| 482 | |
| 483 | requestData, err := json.Marshal(req) |
| 484 | if err != nil { |
| 485 | return nil, err |
| 486 | } |
| 487 | r.mu.RLock() |
| 488 | channel, err := r.conn.Channel() |
| 489 | r.mu.RUnlock() |
| 490 | if err != nil { |
| 491 | return nil, err |
| 492 | } |
| 493 | |
| 494 | consumerID := rid.New(ridConsumerReq) |
| 495 | msgs, err := channel.Consume( |
| 496 | "amq.rabbitmq.reply-to", // queue |
| 497 | consumerID, // consumer |
| 498 | true, // auto-ack |
| 499 | false, // exclusive |
| 500 | false, // no-local |
| 501 | false, // no-wait |
| 502 | nil, // args |
| 503 | ) |
| 504 | if err != nil { |
| 505 | return nil, eris.Wrap(err, "error consuming channel") |
| 506 | } |
| 507 | defer channel.Cancel(consumerID, false) // nolint: errcheck |
| 508 | |
| 509 | ctx, cancel := context.WithTimeout(context.Background(), r.Config.RequestTimeout) |
| 510 | defer cancel() |
| 511 | for i := 0; i <= r.Config.TimeoutRetries; i++ { |
| 512 | err = channel.PublishWithContext( |
| 513 | ctx, |
| 514 | exchangeRequest, // exchange |
| 515 | topic, // routing key |
| 516 | false, // mandatory |
| 517 | false, // immediate |
| 518 | amqp091.Publishing{ |
| 519 | ContentType: "application/json", |
| 520 | CorrelationId: rid.New(ridCorrelation), |
| 521 | Body: requestData, |
| 522 | ReplyTo: "amq.rabbitmq.reply-to", |
| 523 | }) |
| 524 | |
| 525 | if errors.Is(err, context.DeadlineExceeded) { |
| 526 | r.countTimeouts++ |
| 527 | continue |
| 528 | } |
| 529 | } |
| 530 | if err != nil { |
| 531 | return nil, eris.Wrap(err, "failed to publish message") |
| 532 | } |
| 533 | |
| 534 | timer := time.NewTimer(r.Config.RequestTimeout) |
| 535 | defer timer.Stop() |
| 536 | responseCount := 0 |