Как корректно ограничить вызов RPC по таймауту?

Question

[Ромзес Панагиотис]

Используется вызов RPC посредством AMQP (адаптирован с примера в rpc_client.go): после отправки сообщения слушаем, ожидая сообщение с заданным CorrelationId.

Исходный фрагмент кода

resp, err := srv.RPCClient.Call(topic, payload)
if err != nil {
	srv.logger.Error("RPC call error. ", err)
	return err
}

func (c *AMQPClient) Call(topic string, body []byte) (response []byte, err error) {
	// ...
	// publish message with CorrelationId
	// ...

	// Consume messages in loop
	for msg := range msgs {
		if corrID == msg.CorrelationId {
			return msg.Body, nil
		}
	}

	err = errors.New("lost CorrelationId=" + corrID)
	c.logger.Error(err)

	return nil, err
}

Теперь необходимо добавить таймаут операции вызова. Я решил, что необходимо добавить контекст внутрь Call.

Делаю так

ctx, cancel := context.WithTimeout(context.Background(), 1500*time.Millisecond)
defer cancel()

resp, err := srv.RPCClient.Call(ctx, topic, payload)
if err != nil { // должен вернуть ошибку, если таймаут
	srv.logger.Error("RPC call error. ", err)
	return err
}

func (c *AMQPClient) Call(ctx context.Context, topic string, body []byte) (response []byte, err error) {
	// ...
	// publish message with CorrelationId
	// ...

	// Consume messages in loop
	/*
		for msg := range msgs {
			if corrID == msg.CorrelationId {
				return msg.Body, nil
			}
		}
	*/

	for {
		select {
		case msg := <-msgs:
			if corrID == msg.CorrelationId {
				return msg.Body, nil
			}
		case <-ctx.Done():
			return nil, ctx.Err()
		}
	}

	err = errors.New("lost CorrelationId=" + corrID) //  на этой строчке теперь сообщает: unreachable code
	c.logger.Error(err)

	return nil, err
}

Как сделать правильно, избавившись от "unreachable code"?

Answer 1

[Александр Павлюк]

unreachable code можно смело убрать
У вас же и так возвращается ошибка в случае отмены контекста:

case <-ctx.Done():
      return nil, ctx.Err()

Но нужно понимать, что контекст будет по таймауту отменяться только если этот таймаут контексту задан.
Например:

ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
resp, err := srv.RPCClient.Call(ctx, topic, payload)
if err != nil { // тут вернет ошибку "context cancelled" при таймауте
  srv.logger.Error("RPC call error. ", err)
  return err
}

Comments

[Ромзес Панагиотис]

ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()

да, я забыл добавить это в коде в вопросе. Оно уже есть у меня.

То есть,
строка err = errors.New("lost CorrelationId=" + corrID)более вообще неактуальна? Как так?

[Александр Павлюк]

Ромзес Панагиотис, потому что у вас там бесконечный цикл. Он не завершится никогда, возможен только выход из функции через return внутри select.

[Ромзес Панагиотис]

Александр Павлюк, именно поэтому я не уверен, что могу просто ликвидировать эту строку и поэтому спросил. Может, мне нужно сделать как-то по-другому, чтобы выходил из цикла?

[Александр Павлюк]

Ромзес Панагиотис, зачем вам выходить из цикла?
выход из функции вполне себе стандартное и рабочее решение здесь

Answer 2

[Дмитрий Шицков]

Можно изменить код, к примеру, таким образом. А все что после цикла - убрать. У вас сейчас едиснтвенный способ покинуть цикл - return с завершением текущей функции, поэтому весь код после цикла никогда не будет выполнен

case <-ctx.Done():
  return nil, ctx.Err()
}

case <-ctx.Done():
  err = fmt.Errorf("lost CorrelationId=%s %s", corrID, err.Error())
  c.logger.Error(err)
  return nil, ctx.Err()
}

Comments

[Ромзес Панагиотис]

Спасибо