ORE Studio 0.0.4
Loading...
Searching...
No Matches
client.hpp
1/* -*- mode: c++; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*-
2 *
3 * Copyright (C) 2026 Marco Craveiro <marco.craveiro@gmail.com>
4 *
5 * This program is free software; you can redistribute it and/or modify it under
6 * the terms of the GNU General Public License as published by the Free Software
7 * Foundation; either version 3 of the License, or (at your option) any later
8 * version.
9 *
10 * This program is distributed in the hope that it will be useful, but WITHOUT
11 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
12 * FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
13 * details.
14 *
15 * You should have received a copy of the GNU General Public License along with
16 * this program; if not, write to the Free Software Foundation, Inc., 51
17 * Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
18 *
19 */
20#ifndef ORES_NATS_SERVICE_CLIENT_HPP
21#define ORES_NATS_SERVICE_CLIENT_HPP
22
23#include <chrono>
24#include <memory>
25#include <span>
26#include <string>
27#include <string_view>
28#include <unordered_map>
29#include <cstddef>
30#include <boost/asio/awaitable.hpp>
31#include "ores.nats/config/nats_options.hpp"
32#include "ores.nats/domain/message.hpp"
33#include "ores.nats/service/jetstream_admin.hpp"
34#include "ores.nats/service/subscription.hpp"
35
36namespace ores::nats::service {
37
73class client {
74public:
75 explicit client(config::nats_options opts);
76 ~client();
77
78 client(const client&) = delete;
79 client& operator=(const client&) = delete;
80
81 // -------------------------------------------------------------------------
82 // Lifecycle
83 // -------------------------------------------------------------------------
84
90 void connect();
91
98 void disconnect();
99
100 [[nodiscard]] bool is_connected() const noexcept;
101
102 // -------------------------------------------------------------------------
103 // Core pub/sub (ephemeral — no persistence)
104 // -------------------------------------------------------------------------
105
112 void publish(std::string_view subject,
113 std::span<const std::byte> data,
114 std::unordered_map<std::string, std::string> headers = {});
115
123 [[nodiscard]] message request_sync(
124 std::string_view subject,
125 std::span<const std::byte> data,
126 std::unordered_map<std::string, std::string> headers = {},
127 std::chrono::milliseconds timeout = std::chrono::seconds(30));
128
135 [[nodiscard]] boost::asio::awaitable<message> request(
136 std::string_view subject,
137 std::span<const std::byte> data,
138 std::unordered_map<std::string, std::string> headers = {},
139 std::chrono::milliseconds timeout = std::chrono::seconds(30));
140
147 [[nodiscard]] subscription subscribe(std::string_view subject,
148 message_handler handler);
149
159 [[nodiscard]] subscription queue_subscribe(std::string_view subject,
160 std::string_view queue_group,
161 message_handler handler);
162
163 // -------------------------------------------------------------------------
164 // JetStream (durable, persistent)
165 // -------------------------------------------------------------------------
166
174 void js_publish(std::string_view subject,
175 std::span<const std::byte> data,
176 std::unordered_map<std::string, std::string> headers = {});
177
187 [[nodiscard]] subscription js_subscribe(std::string_view subject,
188 std::string_view durable_name,
189 message_handler handler);
190
197 [[nodiscard]] subscription js_queue_subscribe(std::string_view subject,
198 std::string_view durable_name,
199 std::string_view queue_group,
200 message_handler handler);
201
209 [[nodiscard]] jetstream_admin make_admin();
210
211 // -------------------------------------------------------------------------
212 // Shutdown
213 // -------------------------------------------------------------------------
214
222 void drain();
223
224 // -------------------------------------------------------------------------
225 // Subject helpers
226 // -------------------------------------------------------------------------
227
237 [[nodiscard]] std::string make_subject(std::string_view relative) const;
238
239private:
240 struct impl;
241 std::unique_ptr<impl> impl_;
242};
243
244}
245
246#endif
std::function< void(message)> message_handler
Callback type for incoming NATS messages.
Definition message.hpp:72
Configuration for a NATS connection.
Definition nats_options.hpp:30
A received NATS message.
Definition message.hpp:40
NATS client: connection, pub/sub, request/reply, and JetStream.
Definition client.hpp:73
subscription queue_subscribe(std::string_view subject, std::string_view queue_group, message_handler handler)
Queue-group subscribe (competing consumers).
Definition client.cpp:386
void connect()
Connect to the NATS server and initialise JetStream.
Definition client.cpp:227
void js_publish(std::string_view subject, std::span< const std::byte > data, std::unordered_map< std::string, std::string > headers={})
Publish a message to a JetStream stream.
Definition client.cpp:410
jetstream_admin make_admin()
Create a JetStream admin handle for managing streams and consumers.
Definition client.cpp:494
void publish(std::string_view subject, std::span< const std::byte > data, std::unordered_map< std::string, std::string > headers={})
Publish a message to a subject.
Definition client.cpp:298
boost::asio::awaitable< message > request(std::string_view subject, std::span< const std::byte > data, std::unordered_map< std::string, std::string > headers={}, std::chrono::milliseconds timeout=std::chrono::seconds(30))
Asynchronous request/reply (ASIO coroutine).
Definition client.cpp:333
subscription js_queue_subscribe(std::string_view subject, std::string_view durable_name, std::string_view queue_group, message_handler handler)
Queue-group push-subscribe to a JetStream subject.
Definition client.cpp:457
subscription subscribe(std::string_view subject, message_handler handler)
Subscribe to a subject.
Definition client.cpp:368
subscription js_subscribe(std::string_view subject, std::string_view durable_name, message_handler handler)
Push-subscribe to a JetStream subject with a durable consumer.
Definition client.cpp:427
void disconnect()
Disconnect from NATS, freeing all cnats resources.
Definition client.cpp:271
message request_sync(std::string_view subject, std::span< const std::byte > data, std::unordered_map< std::string, std::string > headers={}, std::chrono::milliseconds timeout=std::chrono::seconds(30))
Synchronous request/reply.
Definition client.cpp:310
void drain()
Graceful shutdown.
Definition client.cpp:489
std::string make_subject(std::string_view relative) const
Prepend the configured subject prefix to a relative subject.
Definition client.cpp:290
JetStream management API.
Definition jetstream_admin.hpp:53
RAII handle for a NATS subscription.
Definition subscription.hpp:41