Branch data Line data Source code
1 : : // Copyright 2026 HPActor Contributors
2 : : //
3 : : // Licensed under the Apache License, Version 2.0 (the "License");
4 : : // you may not use this file except in compliance with the License.
5 : : // You may obtain a copy of the License at
6 : : //
7 : : // http://www.apache.org/licenses/LICENSE-2.0
8 : : //
9 : : // Unless required by applicable law or agreed to in writing, software
10 : : // distributed under the License is distributed on an "AS IS" BASIS,
11 : : // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 : : // See the License for the specific language governing permissions and
13 : : // limitations under the License.
14 : :
15 : : // ActorProxy implementation - see actor_proxy.hpp
16 : :
17 : : #include <hpactor/core/actor_system.hpp>
18 : : #include <hpactor/mailbox/dead_letter_queue.hpp>
19 : : #include <hpactor/net/actor_location_cache.hpp>
20 : : #include <hpactor/net/frame.hpp>
21 : : #include <hpactor/net/service_discovery.hpp>
22 : : #include <hpactor/net/transport.hpp>
23 : : #include <hpactor/ref/actor_proxy.hpp>
24 : :
25 : : namespace hpactor {
26 : :
27 : 18 : ActorProxy::ActorProxy(ActorAddress address, net::Transport* transport)
28 : 18 : : address_(address), transport_(transport) {}
29 : :
30 : 3 : ActorProxy::ActorProxy(const ActorAddress& addr, ActorSystem* system)
31 : 3 : : address_(addr),
32 : 3 : transport_(system != nullptr ? system->get_transport_for(addr.endpoint)
33 : : : nullptr),
34 : 3 : system_(system) {}
35 : :
36 : 1 : void ActorProxy::send(const ActorAddress& target, TypedMessage msg) {
37 : 1 : (void)try_send(target, std::move(msg));
38 : 1 : }
39 : :
40 : : mailbox::EnqueueResult
41 : 3 : ActorProxy::try_send(const ActorAddress& target, TypedMessage msg,
42 : : mailbox::DeliveryOptions /*options*/) {
43 : 3 : if (transport_ == nullptr) {
44 : : // Capture dead letter: no transport available
45 : 3 : if (system_) {
46 : 1 : mailbox::DeadLetterRecord dl;
47 : 1 : dl.reason = mailbox::DeadLetterReason::RemoteNodeUnreachable;
48 : 1 : dl.source = mailbox::DeadLetterSource::ActorProxy;
49 : 1 : dl.sender = msg.sender_address().id != ActorId{0}
50 : 1 : ? msg.sender_address()
51 : : : address_;
52 : 1 : dl.target = target;
53 : 1 : dl.type_tag = msg.type_id();
54 : 1 : dl.payload_sample = msg.payload();
55 : 1 : (void)system_->dead_letter(std::move(dl));
56 : 1 : }
57 : 3 : return {mailbox::EnqueueResultCode::ActorNotFound, target.id};
58 : : }
59 : :
60 : : // Resolve via location cache or discovery
61 : 0 : ActorAddress resolved_target = target;
62 : 0 : if (location_cache_) {
63 : 0 : auto cached = location_cache_->get(target.id);
64 : 0 : if (cached) {
65 : 0 : resolved_target.endpoint = *cached;
66 : : }
67 : : }
68 : 0 : if (discovery_) {
69 : 0 : auto* member = discovery_->discover(resolved_target.endpoint);
70 : 0 : if (!member) {
71 : : // Capture dead letter: no route to target
72 : 0 : if (system_) {
73 : 0 : mailbox::DeadLetterRecord dl;
74 : 0 : dl.reason = mailbox::DeadLetterReason::MissingRoute;
75 : 0 : dl.source = mailbox::DeadLetterSource::ServiceDiscovery;
76 : 0 : dl.sender = msg.sender_address().id != ActorId{0}
77 : 0 : ? msg.sender_address()
78 : : : address_;
79 : 0 : dl.target = target;
80 : 0 : dl.type_tag = msg.type_id();
81 : 0 : dl.payload_sample = msg.payload();
82 : 0 : (void)system_->dead_letter(std::move(dl));
83 : 0 : }
84 : 0 : return {mailbox::EnqueueResultCode::ActorNotFound, target.id};
85 : : }
86 : 0 : resolved_target.endpoint = member->identity.endpoint;
87 : 0 : if (location_cache_) {
88 : 0 : location_cache_->put(target.id, resolved_target.endpoint);
89 : : }
90 : : }
91 : :
92 : 0 : net::WireFrame frame;
93 : : // Use msg.sender_address() if present, fall back to the proxy address
94 : : const auto& sender_addr =
95 : 0 : msg.sender_address().id != ActorId{0} ? msg.sender_address() : address_;
96 : 0 : net::to_proto(frame.pb_frame.mutable_sender(), sender_addr);
97 : 0 : net::to_proto(frame.pb_frame.mutable_receiver(), resolved_target);
98 : 0 : frame.pb_frame.set_message_id(generate_message_id().value());
99 : 0 : frame.pb_frame.set_type_tag(static_cast<uint32_t>(msg.type_id()));
100 : 0 : frame.pb_frame.set_payload(reinterpret_cast<const char*>(msg.payload().data()),
101 : 0 : msg.payload().size());
102 : :
103 : 0 : if (msg.has_trace_context()) {
104 : 0 : net::to_proto(frame.pb_frame.mutable_trace_context(), msg.trace_context());
105 : : }
106 : :
107 : 0 : if (!transport_->try_send(resolved_target, frame.encode())) {
108 : : // Capture dead letter: transport refused the message
109 : 0 : if (system_) {
110 : 0 : mailbox::DeadLetterRecord dl;
111 : 0 : dl.reason = mailbox::DeadLetterReason::TransportSendFailed;
112 : 0 : dl.source = mailbox::DeadLetterSource::Transport;
113 : 0 : dl.sender = msg.sender_address().id != ActorId{0}
114 : 0 : ? msg.sender_address()
115 : : : address_;
116 : 0 : dl.target = target;
117 : 0 : dl.type_tag = msg.type_id();
118 : 0 : dl.payload_sample = msg.payload();
119 : 0 : (void)system_->dead_letter(std::move(dl));
120 : 0 : }
121 : 0 : return {mailbox::EnqueueResultCode::Rejected, target.id};
122 : : }
123 : 0 : return {mailbox::EnqueueResultCode::Accepted, target.id};
124 : 0 : }
125 : :
126 : : } // namespace hpactor
|
