rocketmq-client-rust 1.0.0

Rust implementation of Apache rocketmq client
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81

// Copyright 2026 The RocketMQ Rust Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use super::*;
use rocketmq_error::fields;
use rocketmq_error::Error;
use rocketmq_error::ErrorContext;
use rocketmq_transport::api::OutboundRequestContractReason;
use rocketmq_transport::api::OutboundRequestRejectionReason;
use rocketmq_transport::api::OutboundRequestStage;
use rocketmq_transport::api::TransportError;

pub(super) fn direct_invoke_result(
    result: Result<OutboundRequestOutcome, TransportError>,
) -> ClientResult<RemotingCommand> {
    match result {
        Ok(OutboundRequestOutcome::Response(response)) => Ok(response),
        Ok(OutboundRequestOutcome::Rejected(rejection)) => Err(match rejection.reason() {
            OutboundRequestRejectionReason::DeadlineExpired => {
                let mut context = ErrorContext::new().with_text(fields::OPERATION_DIAGNOSTIC, "client_invoke");
                if let Some(timeout_millis) = rejection.timeout_millis() {
                    context = context.with_u64(fields::TIMEOUT_MS, timeout_millis);
                }
                ClientError::from_shared(Arc::new(
                    Error::new(&rocketmq_error::CORE_OPERATION_TIMED_OUT).with_context(context),
                ))
            }
            OutboundRequestRejectionReason::ClientStopping => {
                ClientError::from_shared(Arc::new(Error::new(&rocketmq_error::CLIENT_LIFECYCLE_NOT_STARTED)))
            }
            OutboundRequestRejectionReason::QueueSaturated => ClientError::from_shared(Arc::new(Error::new(
                &rocketmq_error::TRANSPORT_ADMISSION_QUEUE_SATURATED,
            ))),
            OutboundRequestRejectionReason::Cancelled => {
                direct_connection_error(request_stage_phase(rejection.stage()), rejection.remote_addr_present())
            }
            OutboundRequestRejectionReason::SessionClosed => {
                direct_connection_error("closed", rejection.remote_addr_present())
            }
            OutboundRequestRejectionReason::EndpointUnavailable => {
                direct_connection_error("connect", rejection.remote_addr_present())
            }
        }),
        Ok(OutboundRequestOutcome::Contract(contract)) => Err(match contract.reason() {
            OutboundRequestContractReason::NameServerEndpointMissing => {
                direct_connection_error("connect", contract.remote_addr_present())
            }
        }),
        Err(error) => Err(ClientError::from_shared(error.into_shared_error())),
    }
}

fn direct_connection_error(phase: &'static str, remote_addr_present: bool) -> ClientError {
    let mut context = ErrorContext::new().with_text(fields::PHASE, phase);
    if remote_addr_present {
        context = context.with_secret_presence(fields::REMOTE_ADDR_PRESENT);
    }
    ClientError::from_shared(Arc::new(
        Error::new(&rocketmq_error::TRANSPORT_CONNECTION_FAILED).with_context(context),
    ))
}

const fn request_stage_phase(stage: OutboundRequestStage) -> &'static str {
    match stage {
        OutboundRequestStage::BeforeWrite => "before_write",
        OutboundRequestStage::Writing => "writing",
        OutboundRequestStage::AwaitingResponse => "awaiting_response",
        OutboundRequestStage::ResponseReceived => "response_received",
    }
}