Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -132,6 +132,13 @@ jobs:
make install
echo "LIBZMQ=${PWD}/libzmq-build" >> ${GITHUB_ENV}

- name: format
if: ${{ startsWith(matrix.os, 'ubuntu-24') }}
run: |
sudo apt install -y clang-format-20
clang-format-20 --version
git ls-files | grep -E '\.(cpp|hpp)' | xargs clang-format-20 --dry-run --Werror

- name: build
env:
CMAKE_PREFIX_PATH: ${{ env.LIBZMQ }}
Expand Down
5 changes: 5 additions & 0 deletions .pre-commit-config.yaml
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
repos:
- repo: https://github.com/pre-commit/mirrors-clang-format
rev: 'v20.1.8'
hooks:
- id: clang-format
18 changes: 6 additions & 12 deletions examples/multipart_messages.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -7,25 +7,19 @@ int main()
zmq::socket_t sock1(ctx, zmq::socket_type::push);
zmq::socket_t sock2(ctx, zmq::socket_type::pull);
sock1.bind("tcp://127.0.0.1:*");
const std::string last_endpoint =
sock1.get(zmq::sockopt::last_endpoint);
std::cout << "Connecting to "
<< last_endpoint << std::endl;
const std::string last_endpoint = sock1.get(zmq::sockopt::last_endpoint);
std::cout << "Connecting to " << last_endpoint << std::endl;
sock2.connect(last_endpoint);

std::array<zmq::const_buffer, 2> send_msgs = {
zmq::str_buffer("foo"),
zmq::str_buffer("bar!")
};
std::array<zmq::const_buffer, 2> send_msgs = {zmq::str_buffer("foo"),
zmq::str_buffer("bar!")};
if (!zmq::send_multipart(sock1, send_msgs))
return 1;

std::vector<zmq::message_t> recv_msgs;
const auto ret = zmq::recv_multipart(
sock2, std::back_inserter(recv_msgs));
const auto ret = zmq::recv_multipart(sock2, std::back_inserter(recv_msgs));
if (!ret)
return 1;
std::cout << "Got " << *ret
<< " messages" << std::endl;
std::cout << "Got " << *ret << " messages" << std::endl;
return 0;
}
12 changes: 8 additions & 4 deletions examples/pubsub_multithread_inproc.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,8 @@
#include "zmq.hpp"
#include "zmq_addon.hpp"

void PublisherThread(zmq::context_t *ctx) {
void PublisherThread(zmq::context_t *ctx)
{
// Prepare publisher
zmq::socket_t publisher(*ctx, zmq::socket_type::pub);
publisher.bind("inproc://#1");
Expand All @@ -26,7 +27,8 @@ void PublisherThread(zmq::context_t *ctx) {
}
}

void SubscriberThread1(zmq::context_t *ctx) {
void SubscriberThread1(zmq::context_t *ctx)
{
// Prepare subscriber
zmq::socket_t subscriber(*ctx, zmq::socket_type::sub);
subscriber.connect("inproc://#1");
Expand All @@ -48,7 +50,8 @@ void SubscriberThread1(zmq::context_t *ctx) {
}
}

void SubscriberThread2(zmq::context_t *ctx) {
void SubscriberThread2(zmq::context_t *ctx)
{
// Prepare our context and subscriber
zmq::socket_t subscriber(*ctx, zmq::socket_type::sub);
subscriber.connect("inproc://#1");
Expand All @@ -69,7 +72,8 @@ void SubscriberThread2(zmq::context_t *ctx) {
}
}

int main() {
int main()
{
/*
* No I/O threads are involved in passing messages using the inproc transport.
* Therefore, if you are using a ØMQ context for in-process messaging only you
Expand Down
63 changes: 28 additions & 35 deletions tests/active_poller.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@
#include <cstring>

#if !defined(_WIN32)
#include <unistd.h>
#include <unistd.h>
#endif // !_WIN32

TEST_CASE("create destroy", "[active_poller]")
Expand Down Expand Up @@ -95,18 +95,15 @@ TEST_CASE("add fd handler", "[active_poller]")
{
int fd = 1;
zmq::active_poller_t active_poller;
CHECK_NOTHROW(
active_poller.add(fd, zmq::event_flags::pollin, no_op_handler));
CHECK_NOTHROW(active_poller.add(fd, zmq::event_flags::pollin, no_op_handler));
}

TEST_CASE("remove fd handler", "[active_poller]")
{
int fd = 1;
zmq::active_poller_t active_poller;
CHECK_NOTHROW(
active_poller.add(fd, zmq::event_flags::pollin, no_op_handler));
CHECK_NOTHROW(
active_poller.remove(fd));
CHECK_NOTHROW(active_poller.add(fd, zmq::event_flags::pollin, no_op_handler));
CHECK_NOTHROW(active_poller.remove(fd));
CHECK_THROWS_ZMQ_ERROR(EINVAL, active_poller.remove(100));
}

Expand All @@ -117,7 +114,7 @@ TEST_CASE("mixed socket and fd handlers", "[active_poller]")
{
int pipefd[2];
::pipe(pipefd);

zmq::context_t context;
constexpr char inprocSocketAddress[] = "inproc://mixed-handlers";
zmq::socket_t socket_rcv{context, zmq::socket_type::pair};
Expand All @@ -128,31 +125,29 @@ TEST_CASE("mixed socket and fd handlers", "[active_poller]")
unsigned eventsFd = 0;
unsigned eventsSocket = 0;

constexpr char messageText[] = "message";
constexpr size_t messageSize = sizeof(messageText);
constexpr char messageText[] = "message";
constexpr size_t messageSize = sizeof(messageText);

zmq::active_poller_t active_poller;
CHECK_NOTHROW(
active_poller.add(pipefd[0], zmq::event_flags::pollin, [&](zmq::event_flags flags) {
if (flags == zmq::event_flags::pollin)
{
char buffer[256];
CHECK(messageSize == ::read(pipefd[0], buffer, messageSize));
CHECK(0 == std::strcmp(buffer, messageText));
++eventsFd;
}
}));
CHECK_NOTHROW(
active_poller.add(socket_rcv, zmq::event_flags::pollin, [&](zmq::event_flags flags) {
if (flags == zmq::event_flags::pollin)
{
zmq::message_t msg;
CHECK(socket_rcv.recv(msg, zmq::recv_flags::dontwait).has_value());
CHECK(messageSize == msg.size());
CHECK(0 == std::strcmp(messageText, msg.data<const char>()));
++eventsSocket;
}
}));
CHECK_NOTHROW(active_poller.add(
pipefd[0], zmq::event_flags::pollin, [&](zmq::event_flags flags) {
if (flags == zmq::event_flags::pollin) {
char buffer[256];
CHECK(messageSize == ::read(pipefd[0], buffer, messageSize));
CHECK(0 == std::strcmp(buffer, messageText));
++eventsFd;
}
}));
CHECK_NOTHROW(active_poller.add(
socket_rcv, zmq::event_flags::pollin, [&](zmq::event_flags flags) {
if (flags == zmq::event_flags::pollin) {
zmq::message_t msg;
CHECK(socket_rcv.recv(msg, zmq::recv_flags::dontwait).has_value());
CHECK(messageSize == msg.size());
CHECK(0 == std::strcmp(messageText, msg.data<const char>()));
++eventsSocket;
}
}));

// send/rcv socket pair
zmq::message_t msg{messageText, messageSize};
Expand Down Expand Up @@ -369,8 +364,7 @@ TEST_CASE("modify invalid socket throws", "[active_poller]")
zmq::socket_t a{context, zmq::socket_type::push};
zmq::socket_t b{std::move(a)};
zmq::active_poller_t active_poller;
CHECK_THROWS_AS(active_poller.modify(a, zmq::event_flags::pollin),
zmq::error_t);
CHECK_THROWS_AS(active_poller.modify(a, zmq::event_flags::pollin), zmq::error_t);
}

TEST_CASE("modify not added throws", "[active_poller]")
Expand All @@ -380,8 +374,7 @@ TEST_CASE("modify not added throws", "[active_poller]")
zmq::socket_t b{context, zmq::socket_type::push};
zmq::active_poller_t active_poller;
CHECK_NOTHROW(active_poller.add(a, zmq::event_flags::pollin, no_op_handler));
CHECK_THROWS_AS(active_poller.modify(b, zmq::event_flags::pollin),
zmq::error_t);
CHECK_THROWS_AS(active_poller.modify(b, zmq::event_flags::pollin), zmq::error_t);
}

TEST_CASE("modify simple", "[active_poller]")
Expand Down
12 changes: 6 additions & 6 deletions tests/buffer.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -120,21 +120,21 @@ TEST_CASE("mutable_buffer creation C array", "[buffer]")
BT d[10] = {};
zmq::mutable_buffer b = zmq::buffer(d);
CHECK(b.size() == 10 * sizeof(BT));
CHECK(b.data() == static_cast<BT*>(d));
CHECK(b.data() == static_cast<BT *>(d));
zmq::const_buffer b2 = zmq::buffer(d, 4);
CHECK(b2.size() == 4);
CHECK(b2.data() == static_cast<BT*>(d));
CHECK(b2.data() == static_cast<BT *>(d));
}

TEST_CASE("const_buffer creation C array", "[buffer]")
{
const BT d[10] = {};
zmq::const_buffer b = zmq::buffer(d);
CHECK(b.size() == 10 * sizeof(BT));
CHECK(b.data() == static_cast<const BT*>(d));
CHECK(b.data() == static_cast<const BT *>(d));
zmq::const_buffer b2 = zmq::buffer(d, 4);
CHECK(b2.size() == 4);
CHECK(b2.data() == static_cast<const BT*>(d));
CHECK(b2.data() == static_cast<const BT *>(d));
}

TEST_CASE("mutable_buffer creation array", "[buffer]")
Expand Down Expand Up @@ -241,13 +241,13 @@ TEST_CASE("const_buffer creation with str_buffer", "[buffer]")
const wchar_t wd[10] = {};
zmq::const_buffer b = zmq::str_buffer(wd);
CHECK(b.size() == 9 * sizeof(wchar_t));
CHECK(b.data() == static_cast<const wchar_t*>(wd));
CHECK(b.data() == static_cast<const wchar_t *>(wd));

zmq::const_buffer b2_null = zmq::buffer("hello");
constexpr zmq::const_buffer b2 = zmq::str_buffer("hello");
CHECK(b2_null.size() == 6);
CHECK(b2.size() == 5);
CHECK(std::string(static_cast<const char*>(b2.data()), b2.size()) == "hello");
CHECK(std::string(static_cast<const char *>(b2.data()), b2.size()) == "hello");
}

TEST_CASE("const_buffer creation with zbuf string literal char", "[buffer]")
Expand Down
38 changes: 18 additions & 20 deletions tests/codec_multipart.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -10,11 +10,10 @@ TEST_CASE("multipart codec empty", "[codec_multipart]")
multipart_t mmsg;
message_t msg = mmsg.encode();
CHECK(msg.size() == 0);

multipart_t mmsg2;
mmsg2.decode_append(msg);
CHECK(mmsg2.size() == 0);

}

TEST_CASE("multipart codec small", "[codec_multipart]")
Expand Down Expand Up @@ -42,7 +41,7 @@ TEST_CASE("multipart codec big", "[codec_multipart]")
{
using namespace zmq;

message_t big(495); // large size packing
message_t big(495); // large size packing
big.data<char>()[0] = 'X';

multipart_t mmsg;
Expand All @@ -66,10 +65,8 @@ TEST_CASE("multipart codec decode bad data overflow", "[codec_multipart]")
message_t wrong_size(bad_data, 3);
CHECK(wrong_size.size() == 3);
CHECK(wrong_size.data<char>()[0] == 5);

CHECK_THROWS_AS(
multipart_t::decode(wrong_size),
std::out_of_range);

CHECK_THROWS_AS(multipart_t::decode(wrong_size), std::out_of_range);
}

TEST_CASE("multipart codec decode bad data extra data", "[codec_multipart]")
Expand All @@ -80,10 +77,8 @@ TEST_CASE("multipart codec decode bad data extra data", "[codec_multipart]")
message_t wrong_size(bad_data, 3);
CHECK(wrong_size.size() == 3);
CHECK(wrong_size.data<char>()[0] == 1);

CHECK_THROWS_AS(
multipart_t::decode(wrong_size),
std::out_of_range);

CHECK_THROWS_AS(multipart_t::decode(wrong_size), std::out_of_range);
}


Expand All @@ -110,14 +105,15 @@ TEST_CASE("multipart codec encode too big", "[codec_multipart]")
}
#endif

TEST_CASE("multipart codec free function with vector of message_t", "[codec_multipart]")
TEST_CASE("multipart codec free function with vector of message_t",
"[codec_multipart]")
{
using namespace zmq;
std::vector<message_t> parts;
parts.emplace_back("Hello", 5);
parts.emplace_back("World",5);
parts.emplace_back("World", 5);
auto msg = encode(parts);
CHECK(msg.size() == 1 + 5 + 1 + 5 );
CHECK(msg.size() == 1 + 5 + 1 + 5);
CHECK(msg.data<unsigned char>()[0] == 5);
CHECK(msg.data<unsigned char>()[1] == 'H');
CHECK(msg.data<unsigned char>()[6] == 5);
Expand All @@ -130,14 +126,15 @@ TEST_CASE("multipart codec free function with vector of message_t", "[codec_mult
CHECK(parts[1].size() == 5);
}

TEST_CASE("multipart codec free function with vector of const_buffer", "[codec_multipart]")
TEST_CASE("multipart codec free function with vector of const_buffer",
"[codec_multipart]")
{
using namespace zmq;
std::vector<const_buffer> parts;
parts.emplace_back("Hello", 5);
parts.emplace_back("World",5);
parts.emplace_back("World", 5);
auto msg = encode(parts);
CHECK(msg.size() == 1 + 5 + 1 + 5 );
CHECK(msg.size() == 1 + 5 + 1 + 5);
CHECK(msg.data<unsigned char>()[0] == 5);
CHECK(msg.data<unsigned char>()[1] == 'H');
CHECK(msg.data<unsigned char>()[6] == 5);
Expand All @@ -150,16 +147,17 @@ TEST_CASE("multipart codec free function with vector of const_buffer", "[codec_m
CHECK(parts[1].size() == 5);
}

TEST_CASE("multipart codec free function with vector of mutable_buffer", "[codec_multipart]")
TEST_CASE("multipart codec free function with vector of mutable_buffer",
"[codec_multipart]")
{
using namespace zmq;
std::vector<mutable_buffer> parts;
char hello[6] = "Hello";
parts.emplace_back(hello, 5);
char world[6] = "World";
parts.emplace_back(world,5);
parts.emplace_back(world, 5);
auto msg = encode(parts);
CHECK(msg.size() == 1 + 5 + 1 + 5 );
CHECK(msg.size() == 1 + 5 + 1 + 5);
CHECK(msg.data<unsigned char>()[0] == 5);
CHECK(msg.data<unsigned char>()[1] == 'H');
CHECK(msg.data<unsigned char>()[6] == 5);
Expand Down
16 changes: 5 additions & 11 deletions tests/context.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -51,15 +51,13 @@ TEST_CASE("context - use socket after shutdown", "[context]")
zmq::context_t context;
zmq::socket_t sock(context, zmq::socket_type::rep);
context.shutdown();
try
{
try {
sock.connect("inproc://test");
zmq::message_t msg;
(void)sock.recv(msg, zmq::recv_flags::dontwait);
(void) sock.recv(msg, zmq::recv_flags::dontwait);
REQUIRE(false);
}
catch (const zmq::error_t& e)
{
catch (const zmq::error_t &e) {
REQUIRE(e.num() == ETERM);
}
}
Expand All @@ -73,12 +71,8 @@ TEST_CASE("context set/get options", "[context]")
CHECK(context.get(zmq::ctxopt::io_threads) == 5);
#endif

CHECK_THROWS_AS(
context.set(static_cast<zmq::ctxopt>(-42), 5),
zmq::error_t);
CHECK_THROWS_AS(context.set(static_cast<zmq::ctxopt>(-42), 5), zmq::error_t);

CHECK_THROWS_AS(
context.get(static_cast<zmq::ctxopt>(-42)),
zmq::error_t);
CHECK_THROWS_AS(context.get(static_cast<zmq::ctxopt>(-42)), zmq::error_t);
}
#endif
4 changes: 2 additions & 2 deletions tests/message.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -3,9 +3,9 @@

#if defined(ZMQ_CPP11)
static_assert(!std::is_copy_constructible<zmq::message_t>::value,
"message_t should not be copy-constructible");
"message_t should not be copy-constructible");
static_assert(!std::is_copy_assignable<zmq::message_t>::value,
"message_t should not be copy-assignable");
"message_t should not be copy-assignable");
#endif
#if (__cplusplus >= 201703L)
static_assert(std::is_nothrow_swappable<zmq::message_t>::value,
Expand Down
Loading