<|start_filename|>include/nngpp/ctx_view.h<|end_filename|>
#ifndef NNGPP_CTX_VIEW_H
#define NNGPP_CTX_VIEW_H
#include "aio_view.h"

namespace nng {

struct ctx_view {
protected:
	nng_ctx c{0};
	
public:
	ctx_view() = default;

	ctx_view( nng_ctx c ) noexcept : c(c) {}

	nng_ctx get() const noexcept {
		return c;
	}

	explicit operator bool() const noexcept {
		return c.id != 0;
	}
	
	int id() const noexcept {
		return nng_ctx_id(c);
	}
	
	void send( aio_view a ) const noexcept {
		nng_ctx_send(c,a.get());
	}

	void recv( aio_view a ) const noexcept {
		nng_ctx_recv(c,a.get());
	}
	
	void set_opt( const char* name, view v ) const {
		int r = nng_ctx_set(c,name,v.data(),v.size());
		if( r != 0 ) {
			throw exception(r,"nng_ctx_set");
		}
	}
	
	void set_opt_bool( const char* name, bool value ) const {
		int r = nng_ctx_set_bool(c,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_ctx_set_bool");
		}
	}
	
	void set_opt_int( const char* name, int value ) const {
		int r = nng_ctx_set_int(c,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_ctx_set_int");
		}
	}
	
	void set_opt_size( const char* name, size_t value ) const {
		int r = nng_ctx_set_size(c,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_ctx_set_size");
		}
	}
	
	void set_opt_uint64( const char* name, uint64_t value ) const {
		int r = nng_ctx_set_uint64(c,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_ctx_set_uint64");
		}
	}
	
	void set_opt_string( const char* name, const char* value ) const {
		int r = nng_ctx_set_string(c,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_ctx_set_string");
		}
	}
	
	void set_opt_ptr( const char* name, void* value ) const {
		int r = nng_ctx_set_ptr(c,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_ctx_set_ptr");
		}
	}
	
	void set_opt_ms( const char* name, nng_duration value ) const {
		int r = nng_ctx_set_ms(c,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_ctx_set_ms");
		}
	}
	
	void set_opt_addr( const char* name, const nng_sockaddr* value ) const {
		int r = nng_ctx_set_addr(c,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_ctx_set_addr");
		}
	}
	
	size_t get_opt( const char* name, buffer& out ) const {
		size_t size = out.size();
		int r = nng_ctx_get(c,name,out.data(),&size);
		if( r != 0 ) {
			throw exception(r,"nng_ctx_get");
		}
		return size;
	}
	
	bool get_opt_bool( const char* name ) const {
		bool out;
		int r = nng_ctx_get_bool(c,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_ctx_get_bool");
		}
		return out;
	}
	
	int get_opt_int( const char* name ) const {
		int out;
		int r = nng_ctx_get_int(c,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_ctx_get_int");
		}
		return out;
	}
	
	size_t get_opt_size( const char* name ) const {
		size_t out;
		int r = nng_ctx_get_size(c,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_ctx_get_size");
		}
		return out;
	}
	
	uint64_t get_opt_uint64( const char* name ) const {
		uint64_t out;
		int r = nng_ctx_get_uint64(c,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_ctx_get_uint64");
		}
		return out;
	}
	
	buffer get_opt_string( const char* name ) const {
		char* out;
		int r = nng_ctx_get_string(c,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_ctx_get_string");
		}
		return buffer(out, strlen(out) + 1);
	}
	
	void* get_opt_ptr( const char* name ) const {
		void* out;
		int r = nng_ctx_get_ptr(c,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_ctx_get_ptr");
		}
		return out;
	}
	
	nng_duration get_opt_ms( const char* name ) const {
		nng_duration out;
		int r = nng_ctx_get_ms(c,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_ctx_get_ms");
		}
		return out;
	}
	
	nng_sockaddr get_opt_addr( const char* name ) const {
		nng_sockaddr out;
		int r = nng_ctx_get_addr(c,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_ctx_get_addr");
		}
		return out;
	}
	
};

inline nng_duration get_opt_recv_timeout( ctx_view s ) {
	return s.get_opt_ms( to_name(option::recv_timeout) );
}

inline nng_duration get_opt_send_timeout( ctx_view s ) {
	return s.get_opt_ms( to_name(option::send_timeout) );
}

inline void set_opt_recv_timeout( ctx_view s, nng_duration v ) {
	s.set_opt_ms( to_name(option::recv_timeout), v );
}

inline void set_opt_send_timeout( ctx_view s, nng_duration v ) {
	s.set_opt_ms( to_name(option::send_timeout), v );
}

}

#endif


<|start_filename|>perf/inproc.h<|end_filename|>
#ifndef NNGPP_PERF_INPROC_H
#define NNGPP_PERF_INPROC_H

struct inproc_args {
	int count;
	int msgsize;
	const char* addr;
	void (*func)(const char*, size_t, int);
};

static void do_inproc(void* args) {
	auto ia = (inproc_args*)args;
	ia->func(ia->addr, ia->msgsize, ia->count);
}

#endif


<|start_filename|>include/nngpp/http/res_view.h<|end_filename|>
#ifndef NNGPP_HTTP_RES_VIEW_H
#define NNGPP_HTTP_RES_VIEW_H
#include <nngpp/view.h>
#include "misc.h"

namespace nng { namespace http {

struct res_view {
protected:
	nng_http_res* s = nullptr;
	
public:
	res_view() = default;

	res_view( nng_http_res* s ) noexcept : s(s) {}
	
	nng_http_res* get() const noexcept {
		return s;
	}

	nng_http_res* operator->() const noexcept {
		return s;
	}

	explicit operator bool() const noexcept {
		return s != nullptr;
	}
	
	void reset() const noexcept {
		nng_http_res_reset(s);
	}

	status get_status() const noexcept {
		return (status)nng_http_res_get_status(s);
	}
	
	void set_status( status status ) const {
		int r = nng_http_res_set_status(s,(uint16_t)status);
		if( r != 0 ) {
			throw exception(r,"nng_http_res_set_status");
		}
	}
	
	const char* get_reason() const noexcept {
		return nng_http_res_get_reason(s);
	}
	
	void set_reason( const char* reason ) const {
		int r = nng_http_res_set_reason(s,reason);
		if( r != 0 ) {
			throw exception(r,"nng_http_res_set_reason");
		}
	}
	
	const char* get_header( const char* name ) const noexcept {
		return nng_http_res_get_header(s,name);
	}
	
	void set_header( const char* name, const char* value ) const {
		int r = nng_http_res_set_header(s,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_http_res_set_header");
		}
	}
	
	void add_header( const char* name, const char* value ) const {
		int r = nng_http_res_add_header(s,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_http_res_add_header");
		}
	}
	
	void del_header( const char* name ) const {
		int r = nng_http_res_del_header(s,name);
		if( r != 0 ) {
			throw exception(r,"nng_http_res_del_header");
		}
	}
	
	const char* get_version() const noexcept {
		return nng_http_res_get_version(s);
	}
	
	void set_version( const char* version ) const {
		int r = nng_http_res_set_version(s,version);
		if( r != 0 ) {
			throw exception(r,"nng_http_res_set_version");
		}
	}

	view get_data() const noexcept {
		void* data;
		size_t size;
		nng_http_res_get_data(s,&data,&size);
		return view(data,size);
	}

	void set_data( view v ) const {
		int r = nng_http_res_set_data(s,v.data(),v.size());
		if( r != 0 ) {
			throw exception(r,"nng_http_res_set_data");
		}
	}
	
	void copy_data( view v ) const {
		int r = nng_http_res_copy_data(s,v.data(),v.size());
		if( r != 0 ) {
			throw exception(r,"nng_http_res_copy_data");
		}
	}
};

}}

#endif


<|start_filename|>test/pipe.cpp<|end_filename|>
#include <catch2/catch.hpp>
#include <nngpp/nngpp.h>
#include <nngpp/protocol/pull0.h>
#include <nngpp/protocol/push0.h>
#include <nngpp/platform/platform.h>
#include <mutex>

namespace {

struct testcase {
	nng::mtx lk = nng::make_mtx();
	nng::cv cv{ lk };
	nng::socket s;
	nng::dialer d;
	nng::listener l;
	nng::pipe_view p;
	int add_pre = 0;
	int add_post = 0;
	int rem = 0;
	int err = 0;
	int rej = 0;

	testcase( nng::socket&& s ) : s(std::move(s)) {}
};

bool expect(testcase* t, int* vp, int v) {
	nng_time when = nng_clock() + 5000; // five seconds
	
	std::lock_guard<nng::mtx> lock( t->lk );
	while(*vp != v) {
		if( !t->cv.wait_until(when) ) {
			break;
		}
	}
	bool ok = (*vp == v);
	if(!ok) {
		printf("Expected %d but got %d\n", v, *vp);
	}
	return ok;
}

void notify(nng_pipe pp, nng_pipe_ev act, void* arg) {
	 auto t = (testcase*)arg;
	 nng::pipe_view p = pp;
	 
	std::lock_guard<nng::mtx> lock( t->lk );
	if(
		(p.get_socket().id() != t->s.id()) ||
	    (p.get_listener().id() != t->l.id()) ||
	    (p.get_dialer().id() != t->d.id())
	) {
		t->err++;
		t->cv.wake_all();
		return;
	}
	if (t->add_post > t->add_pre) {
		t->err++;
	}
	switch(act) {
	case NNG_PIPE_EV_ADD_PRE:
		t->add_pre++;
		break;
	case NNG_PIPE_EV_ADD_POST:
		t->add_post++;
		break;
	case NNG_PIPE_EV_REM_POST:
		t->rem++;
		break;
	default:
		t->err++;
		t->cv.wake_all();
		return;
	}
	t->p = p;
	t->cv.wake_all();
}

void reject(nng_pipe pp, nng_pipe_ev act, void* arg) {
	auto t = (testcase*)arg;
	notify(pp, act, arg);
	
	std::lock_guard<nng::mtx> lock( t->lk );
	if(!t->rej) {
		{
			nng::pipe pipe(pp);
		}
		t->rej++;
	}
}

}

TEST_CASE("Pipe notify works","[pipe]") {

	SECTION("We can create a pipeline") {
		testcase push = nng::push::open();
		testcase pull = nng::pull::open();
		auto addr = "inproc://pipe";

		REQUIRE_NOTHROW( set_opt_reconnect_time_min(push.s,10) );
		REQUIRE_NOTHROW( set_opt_reconnect_time_max(push.s,10) );
		REQUIRE_NOTHROW( set_opt_reconnect_time_min(pull.s,10) );
		REQUIRE_NOTHROW( set_opt_reconnect_time_max(pull.s,10) );

		REQUIRE_NOTHROW( push.s.pipe_notify( nng::pipe_ev::add_pre, notify, &push ) );
		REQUIRE_NOTHROW( push.s.pipe_notify( nng::pipe_ev::add_post, notify, &push ) );
		REQUIRE_NOTHROW( push.s.pipe_notify( nng::pipe_ev::rem_post, notify, &push ) );
		REQUIRE_NOTHROW( pull.s.pipe_notify( nng::pipe_ev::add_pre, notify, &pull ) );
		REQUIRE_NOTHROW( pull.s.pipe_notify( nng::pipe_ev::add_post, notify, &pull ) );
		REQUIRE_NOTHROW( pull.s.pipe_notify( nng::pipe_ev::rem_post, notify, &pull ) );

		SECTION("Dialing works") {
			REQUIRE_NOTHROW( pull.l = nng::listener(pull.s,addr) );
			REQUIRE_NOTHROW( push.d = nng::dialer(push.s,addr) );
			
			REQUIRE(pull.l);
			REQUIRE(push.d);

			REQUIRE_NOTHROW( set_opt_reconnect_time_min(push.d,10) );
			REQUIRE_NOTHROW( set_opt_reconnect_time_max(push.d,10) );

			REQUIRE_NOTHROW( pull.l.start() );
			REQUIRE_NOTHROW( push.d.start() );

			REQUIRE(expect(&pull, &pull.add_pre, 1));
			REQUIRE(expect(&pull, &pull.add_post, 1));
			REQUIRE(expect(&pull, &pull.add_pre, 1));
			REQUIRE(expect(&pull, &pull.add_post, 1));
			REQUIRE(expect(&pull, &pull.rem, 0));
			REQUIRE(expect(&pull, &pull.err, 0));
			REQUIRE(expect(&push, &push.add_pre, 1));
			REQUIRE(expect(&push, &push.add_post, 1));
			REQUIRE(expect(&push, &push.rem, 0));
			REQUIRE(expect(&push, &push.err, 0));

			SECTION("We can send a frame") {
				auto msg = nng::make_msg(0);
				msg.body().append("hello");
				push.s.send( std::move(msg) );
				msg = pull.s.recv_msg();
				REQUIRE(msg);
				REQUIRE( msg.body().get() == "hello" );
				REQUIRE( msg.get_pipe().id() == pull.p.id() );
			}

			SECTION("Reconnection works") {
				REQUIRE(expect(&pull, &pull.add_pre, 1));
				REQUIRE(expect(&pull, &pull.add_post, 1));
				nng_pipe_close(pull.p.get());

				REQUIRE(expect(&pull, &pull.rem, 1));
				REQUIRE(expect(&pull, &pull.err, 0));
				REQUIRE(expect(&pull, &pull.add_pre, 2));
				REQUIRE(expect(&pull, &pull.add_post, 2));

				REQUIRE(expect(&push, &push.rem, 1));
				REQUIRE(expect(&push, &push.err, 0));
				REQUIRE(expect(&push, &push.add_pre, 2));
				REQUIRE(expect(&push, &push.add_post, 2));

				SECTION("They still exchange frames") {
					nng::msleep(200);
					nng::pipe_view p1;
					{
						auto msg = nng::make_msg(0);
						msg.body().append("hello");
						push.s.send( std::move(msg) );
						msg = pull.s.recv_msg();
						REQUIRE(msg);
						REQUIRE( msg.body().get() == "hello" );
						REQUIRE( msg.get_pipe().id() == pull.p.id() );

						p1 = msg.get_pipe();
					}
					REQUIRE(p1.id() == pull.p.id());
				}
			}
		}

		SECTION("Reject works") {
			
			REQUIRE_NOTHROW( pull.s.pipe_notify( nng::pipe_ev::add_pre, reject, &pull ) );

			REQUIRE_NOTHROW( pull.l = nng::listener(pull.s,addr) );
			REQUIRE_NOTHROW( push.d = nng::dialer(push.s,addr) );
			
			REQUIRE(pull.l);
			REQUIRE(push.d);
			
			REQUIRE_NOTHROW( pull.l.start() );
			nng::msleep(100);
			REQUIRE_NOTHROW( push.d.start() );
			
			REQUIRE(expect(&pull, &pull.add_pre, 2));
			REQUIRE(expect(&pull, &pull.add_post, 1));
			REQUIRE(expect(&pull, &pull.rem, 1));
			REQUIRE(expect(&pull, &pull.err, 0));
			REQUIRE(expect(&push, &push.add_pre, 2));
			REQUIRE(expect(&push, &push.add_post, 2));
			REQUIRE(expect(&push, &push.rem, 1));
			REQUIRE(expect(&push, &push.err, 0));
		}
	}
}


<|start_filename|>include/nngpp/platform/platform.h<|end_filename|>
#ifndef NNGPP_PLATFORM_H
#define NNGPP_PLATFORM_H
#include "cv.h"
#include "mtx.h"
#include "thread.h"

namespace nng {

inline nng_time clock() noexcept {
	return nng_clock();
}

inline void msleep( nng_duration dt ) noexcept {
	nng_msleep(dt);
}

inline uint32_t random() noexcept {
	return nng_random();
}

}

#endif


<|start_filename|>include/nngpp/listener_view.h<|end_filename|>
#ifndef NNGPP_LISTENER_VIEW_H
#define NNGPP_LISTENER_VIEW_H
#include "option.h"
#include "buffer.h"

namespace nng {

struct listener_view {
protected:
	nng_listener lis{0};
	
public:
	listener_view() = default;
	
	listener_view( nng_listener lis ) noexcept : lis(lis) {}
	
	nng_listener get() const noexcept {
		return lis;
	}
	
	explicit operator bool() const noexcept {
		return lis.id != 0;
	}
	
	void start( int flags = 0 ) const {
		int r = nng_listener_start(lis,flags);
		if( r != 0 ) {
			throw exception(r,"nng_listener_start");
		}
	}
	
	int id() const noexcept {
		return nng_listener_id(lis);
	}
	
	void set_opt( const char* name, view v ) const {
		int r = nng_listener_set(lis,name,v.data(),v.size());
		if( r != 0 ) {
			throw exception(r,"nng_listener_set");
		}
	}
	
	void set_opt_bool( const char* name, bool value ) const {
		int r = nng_listener_set_bool(lis,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_listener_set_bool");
		}
	}
	
	void set_opt_int( const char* name, int value ) const {
		int r = nng_listener_set_int(lis,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_listener_set_int");
		}
	}
	
	void set_opt_size( const char* name, size_t value ) const {
		int r = nng_listener_set_size(lis,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_listener_set_size");
		}
	}
	
	void set_opt_uint64( const char* name, uint64_t value ) const {
		int r = nng_listener_set_uint64(lis,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_listener_set_uint64");
		}
	}
	
	void set_opt_string( const char* name, const char* value ) const {
		int r = nng_listener_set_string(lis,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_listener_set_string");
		}
	}
	
	void set_opt_ptr( const char* name, void* value ) const {
		int r = nng_listener_set_ptr(lis,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_listener_set_ptr");
		}
	}
	
	void set_opt_ms( const char* name, nng_duration value ) const {
		int r = nng_listener_set_ms(lis,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_listener_set_ms");
		}
	}
	
	void set_opt_addr( const char* name, const nng_sockaddr* value ) const {
		int r = nng_listener_set_addr(lis,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_listener_set_addr");
		}
	}
	
	size_t get_opt( const char* name, buffer& out ) const {
		size_t size = out.size();
		int r = nng_listener_get(lis,name,out.data(),&size);
		if( r != 0 ) {
			throw exception(r,"nng_listener_get");
		}
		return size;
	}

	bool get_opt_bool( const char* name ) const {
		bool out;
		int r = nng_listener_get_bool(lis,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_listener_get_bool");
		}
		return out;
	}
	
	int get_opt_int( const char* name ) const {
		int out;
		int r = nng_listener_get_int(lis,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_listener_get_int");
		}
		return out;
	}
	
	size_t get_opt_size( const char* name ) const {
		size_t out;
		int r = nng_listener_get_size(lis,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_listener_get_size");
		}
		return out;
	}

	uint64_t get_opt_uint64( const char* name ) const {
		uint64_t out;
		int r = nng_listener_get_uint64(lis,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_listener_get_uint64");
		}
		return out;
	}
	
	buffer get_opt_string( const char* name ) const {
		char* data;
		int r = nng_listener_get_string(lis,name,&data);
		if( r != 0 ) {
			throw exception(r,"nng_listener_get_string");
		}
		return buffer(data,strlen(data)+1);
	}
	
	void* get_opt_ptr( const char* name ) const {
		void* out;
		int r = nng_listener_get_ptr(lis,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_listener_get_ptr");
		}
		return out;
	}
	
	nng_duration get_opt_ms( const char* name ) const {
		nng_duration out;
		int r = nng_listener_get_ms(lis,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_listener_get_ms");
		}
		return out;
	}
	
	nng_sockaddr get_opt_addr( const char* name ) const {
		nng_sockaddr out;
		int r = nng_listener_get_addr(lis,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_listener_get_addr");
		}
		return out;
	}
};

inline nng_sockaddr get_opt_local_address( listener_view s ) {
	return s.get_opt_addr( to_name(option::local_address) );
}

inline buffer get_opt_url( listener_view s ) {
	return s.get_opt_string( to_name(option::url) );
}

inline size_t get_opt_recv_size_max( listener_view s ) {
	return s.get_opt_size( to_name(option::recv_size_max) );
}

inline void set_opt_recv_size_max( listener_view s, size_t v ) {
	s.set_opt_size( to_name(option::recv_size_max), v );
}

}

#endif


<|start_filename|>perf/throughput.h<|end_filename|>
#ifndef NNGPP_PERF_THROUGHPUT_H
#define NNGPP_PERF_THROUGHPUT_H
#include <nngpp/nngpp.h>
#include <nngpp/platform/platform.h>
#include <nngpp/protocol/pair1.h>
#include "misc.h"

void throughput_server(const char* addr, size_t msgsize, int count) {
	nng_time start, end;
	{
		auto s = nng::pair::open();
		nng::set_opt_recv_buffer( s, 128 );
		s.listen(addr);

		// Receive first synchronization message.
		s.recv_msg();
		start = nng::clock();

		for(int i = 0; i < count; ++i) {
			auto msg = s.recv_msg();
			if(msg.body().size() != msgsize) {
				die("wrong message size: %d != %d", msg.body().size(), msgsize);
			}
		}
		end = nng::clock();
		// Send a synchronization message (empty) to the other side,
		// and wait a bit to make sure it goes out the wire.
		s.send( nng::view("",0) );
		nng::msleep(200);
	}
	auto total = (float)(end - start) / 1000;
	auto msgpersec = (float)count / total;
	auto mbps = (float)(msgpersec * 8 * msgsize) / (1024 * 1024);
	printf("total time: %.3f [s]\n", total);
	printf("message size: %d [B]\n", (int)msgsize);
	printf("message count: %d\n", count);
	printf("throughput: %.f [msg/s]\n", msgpersec);
	printf("throughput: %.3f [Mb/s]\n", mbps);
}

void throughput_client(const char* addr, size_t msgsize, int count) {
	auto s = nng::pair::open();
	nng::set_opt_send_buffer( s, 128 );
	nng::set_opt_recv_timeout( s, 5000 );
	s.dial(addr);

	// We send one extra zero length message to start the timer.
	s.send( nng::make_msg(0) );

	for(int i = 0; i < count; ++i) {
		s.send( nng::make_msg(msgsize) );
	}
	// Attempt to get the completion indication from the other side.
	s.recv_msg();
}

#endif


<|start_filename|>include/nngpp/stream/stream_view.h<|end_filename|>
#ifndef NNGPP_STREAM_STREAM_VIEW_H
#define NNGPP_STREAM_STREAM_VIEW_H
#include <nngpp/aio_view.h>

namespace nng { namespace stream {

struct stream_view {
protected:
	nng_stream* s = nullptr;
	
public:
	stream_view() = default;
	
	stream_view( nng_stream* s ) noexcept : s(s) {}

	nng_stream* get() const noexcept {
		return s;
	}

	nng_stream* operator->() const noexcept {
		return s;
	}

	explicit operator bool() const noexcept {
		return s != nullptr;
	}

	void close() const noexcept {
		nng_stream_close(s);
	}
	
	void send( nng::aio_view a ) const noexcept {
		nng_stream_send(s,a.get());
	}
	
	void recv( nng::aio_view a ) const noexcept {
		nng_stream_recv(s,a.get());
	}

	void get( const char* name, void* val, size_t* szp ) const {
		int r = nng_stream_get(s,name,val,szp);
		if( r != 0 ) {
			throw exception(r,"nng_stream_get");
		}
	}
	
	template<typename T>
	T get( const char* name ) const {
		T out;
		size_t size = sizeof(out);
		get(name,&out,&size);
		return out;
	}

	bool get_bool( const char* name ) const {
		bool out;
		int r = nng_stream_get_bool(s,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_stream_get_bool");
		}
		return out;
	}
	
	int get_int( const char* name ) const {
		int out;
		int r = nng_stream_get_int(s,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_stream_get_int");
		}
		return out;
	}
	
	size_t get_size( const char* name ) const {
		size_t out;
		int r = nng_stream_get_size(s,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_stream_get_size");
		}
		return out;
	}
	
	uint64_t get_uint64( const char* name ) const {
		uint64_t out;
		int r = nng_stream_get_uint64(s,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_stream_get_uint64");
		}
		return out;
	}

	buffer get_string( const char* name ) const {
		char* data;
		int r = nng_stream_get_string(s,name,&data);
		if( r != 0 ) {
			throw exception(r,"nng_stream_get_string");
		}
		return buffer(data,strlen(data)+1);
	}

	void* get_ptr( const char* name ) const {
		void* out;
		int r = nng_stream_get_ptr(s,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_stream_get_ptr");
		}
		return out;
	}
	
	nng_duration get_ms( const char* name ) const {
		nng_duration out;
		int r = nng_stream_get_ms(s,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_stream_get_ms");
		}
		return out;
	}
	
	nng_sockaddr get_addr( const char* name ) const {
		nng_sockaddr data;
		int r = nng_stream_get_addr(s,name,&data);
		if( r != 0 ) {
			throw exception(r,"nng_stream_get_addr");
		}
		return data;
	}
	
	void set( const char* name, const void* val, size_t sz ) const {
		int r = nng_stream_set(s,name,val,sz);
		if( r != 0 ) {
			throw exception(r,"nng_stream_set");
		}
	}
	
	template<typename T>
	void set( const char* name, const T& v ) const {
		set(name,&v,sizeof(v));
	}

	void set_bool( const char* name, bool value ) const {
		int r = nng_stream_set_bool(s,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_stream_set_bool");
		}
	}
	
	void set_int( const char* name, int value ) const {
		int r = nng_stream_set_int(s,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_stream_set_int");
		}
	}
	
	void set_size( const char* name, size_t value ) const {
		int r = nng_stream_set_size(s,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_stream_set_size");
		}
	}
	
	void set_uint64( const char* name, uint64_t value ) const {
		int r = nng_stream_set_uint64(s,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_stream_set_uint64");
		}
	}
	
	void set_string( const char* name, const char* value ) const {
		int r = nng_stream_set_string(s,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_stream_set_string");
		}
	}
	
	void set_ptr( const char* name, void* value ) const {
		int r = nng_stream_set_ptr(s,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_stream_set_ptr");
		}
	}
	
	void set_ms( const char* name, nng_duration value ) const {
		int r = nng_stream_set_ms(s,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_stream_set_ms");
		}
	}
	
	void set_addr( const char* name, const nng_sockaddr& value ) const {
		int r = nng_stream_set_addr(s,name,&value);
		if( r != 0 ) {
			throw exception(r,"nng_stream_set_addr");
		}
	}

};

}}

#endif


<|start_filename|>include/nngpp/http/misc.h<|end_filename|>
#ifndef NNGPP_HTTP_MISC_H
#define NNGPP_HTTP_MISC_H
#include <nng/supplemental/http/http.h>

namespace nng { namespace http {

enum class status {
	continue_                = NNG_HTTP_STATUS_CONTINUE,
	switching                = NNG_HTTP_STATUS_SWITCHING,
	processing               = NNG_HTTP_STATUS_PROCESSING,
	ok                       = NNG_HTTP_STATUS_OK,
	created                  = NNG_HTTP_STATUS_CREATED,
	accepted                 = NNG_HTTP_STATUS_ACCEPTED,
	not_authoritative        = NNG_HTTP_STATUS_NOT_AUTHORITATIVE,
	no_content               = NNG_HTTP_STATUS_NO_CONTENT,
	reset_content            = NNG_HTTP_STATUS_RESET_CONTENT,
	partial_content          = NNG_HTTP_STATUS_PARTIAL_CONTENT,
	multi_status             = NNG_HTTP_STATUS_MULTI_STATUS,
	already_reported         = NNG_HTTP_STATUS_ALREADY_REPORTED,
	im_used                  = NNG_HTTP_STATUS_IM_USED,
	multiple_choices         = NNG_HTTP_STATUS_MULTIPLE_CHOICES,
	moved_permanently        = NNG_HTTP_STATUS_STATUS_MOVED_PERMANENTLY,
	found                    = NNG_HTTP_STATUS_FOUND,
	see_other                = NNG_HTTP_STATUS_SEE_OTHER,
	not_modified             = NNG_HTTP_STATUS_NOT_MODIFIED,
	use_proxy                = NNG_HTTP_STATUS_USE_PROXY,
	temporary_redirect       = NNG_HTTP_STATUS_TEMPORARY_REDIRECT,
	permanent_redirect       = NNG_HTTP_STATUS_PERMANENT_REDIRECT,
	bad_request              = NNG_HTTP_STATUS_BAD_REQUEST,
	unauthorized             = NNG_HTTP_STATUS_UNAUTHORIZED,
	payment_required         = NNG_HTTP_STATUS_PAYMENT_REQUIRED,
	forbidden                = NNG_HTTP_STATUS_FORBIDDEN,
	not_found                = NNG_HTTP_STATUS_NOT_FOUND,
	method_not_allowed       = NNG_HTTP_STATUS_METHOD_NOT_ALLOWED,
	not_acceptable           = NNG_HTTP_STATUS_NOT_ACCEPTABLE,
	proxy_auth_required      = NNG_HTTP_STATUS_PROXY_AUTH_REQUIRED,
	request_timeout          = NNG_HTTP_STATUS_REQUEST_TIMEOUT,
	conflict                 = NNG_HTTP_STATUS_CONFLICT,
	gone                     = NNG_HTTP_STATUS_GONE,
	length_required          = NNG_HTTP_STATUS_LENGTH_REQUIRED,
	precondition_failed      = NNG_HTTP_STATUS_PRECONDITION_FAILED,
	payload_too_large        = NNG_HTTP_STATUS_PAYLOAD_TOO_LARGE,
	entity_too_long          = NNG_HTTP_STATUS_ENTITY_TOO_LONG,
	unsupported_media_type   = NNG_HTTP_STATUS_UNSUPPORTED_MEDIA_TYPE,
	range_not_satisfiable    = NNG_HTTP_STATUS_RANGE_NOT_SATISFIABLE,
	expectation_failed       = NNG_HTTP_STATUS_EXPECTATION_FAILED,
	teapot                   = NNG_HTTP_STATUS_TEAPOT,
	unprocessable_entity     = NNG_HTTP_STATUS_UNPROCESSABLE_ENTITY,
	locked                   = NNG_HTTP_STATUS_LOCKED,
	failed_dependency        = NNG_HTTP_STATUS_FAILED_DEPENDENCY,
	upgrade_required         = NNG_HTTP_STATUS_UPGRADE_REQUIRED,
	precondition_required    = NNG_HTTP_STATUS_PRECONDITION_REQUIRED,
	too_many_requests        = NNG_HTTP_STATUS_TOO_MANY_REQUESTS,
	headers_too_large        = NNG_HTTP_STATUS_HEADERS_TOO_LARGE,
	unavail_legal_reasons    = NNG_HTTP_STATUS_UNAVAIL_LEGAL_REASONS,
	internal_server_error    = NNG_HTTP_STATUS_INTERNAL_SERVER_ERROR,
	not_implemented          = NNG_HTTP_STATUS_NOT_IMPLEMENTED,
	bad_gateway              = NNG_HTTP_STATUS_BAD_GATEWAY,
	service_unavailable      = NNG_HTTP_STATUS_SERVICE_UNAVAILABLE,
	gateway_timeout          = NNG_HTTP_STATUS_GATEWAY_TIMEOUT,
	http_version_not_supp    = NNG_HTTP_STATUS_HTTP_VERSION_NOT_SUPP,
	variant_also_negotiates  = NNG_HTTP_STATUS_VARIANT_ALSO_NEGOTIATES,
	insufficient_storage     = NNG_HTTP_STATUS_INSUFFICIENT_STORAGE,
	loop_detected            = NNG_HTTP_STATUS_LOOP_DETECTED,
	not_extended             = NNG_HTTP_STATUS_NOT_EXTENDED,
	network_auth_required    = NNG_HTTP_STATUS_NETWORK_AUTH_REQUIRED
};

enum class verb {
	get,
	head,
	post,
	put,
	delete_,
	trace,
	options,
	connect,
	patch
};

namespace detail {

#if __cpp_inline_variables >= 201606
inline
#else
static
#endif
const char* const verb_name[] {
	"GET",
	"HEAD",
	"POST",
	"PUT",
	"DELETE",
	"TRACE",
	"OPTIONS",
	"CONNECT",
	"PATCH"
};

}

inline const char* to_name( verb v ) noexcept {
	return detail::verb_name[(int)v];
}

}}

#endif


<|start_filename|>include/nngpp/transport/zerotier.h<|end_filename|>
#ifndef NNGPP_TRANSPORT_ZEROTIER_H
#define NNGPP_TRANSPORT_ZEROTIER_H
#include <nngpp/listener_view.h>
#include <nngpp/dialer_view.h>
#include <nngpp/pipe_view.h>
#include <nng/transport/zerotier/zerotier.h>

namespace nng { namespace zt {

inline void register_transport() {
	int r = nng_zt_register();
	if( r != 0 ) {
		throw exception(r,"nng_zt_register");
	}
}

enum class status {
	up        = NNG_ZT_STATUS_UP,
	config    = NNG_ZT_STATUS_CONFIG,
	denied    = NNG_ZT_STATUS_DENIED,
	not_found = NNG_ZT_STATUS_NOTFOUND,
	error     = NNG_ZT_STATUS_ERROR,
	obsolete  = NNG_ZT_STATUS_OBSOLETE,
	unknown   = NNG_ZT_STATUS_UNKNOWN
};

enum class option {
	mtu,            // size
	nwid,           // u64
	node,           // u64
	home,           // string
	network_status, // int
	network_name,   // string
	ping_time,      // ms
	ping_tries,     // int
	conn_time,      // ms
	conn_tries,     // int
	orbit,          // u64
	deorbit         // u64
};

namespace detail {

#if __cpp_inline_variables >= 201606
inline
#else
static
#endif
const char* const option_names[] = {
	NNG_OPT_ZT_MTU,
	NNG_OPT_ZT_NWID,
	NNG_OPT_ZT_NODE,
	NNG_OPT_ZT_HOME,
	NNG_OPT_ZT_NETWORK_STATUS,
	NNG_OPT_ZT_NETWORK_NAME,
	NNG_OPT_ZT_PING_TIME,
	NNG_OPT_ZT_PING_TRIES,
	NNG_OPT_ZT_CONN_TIME,
	NNG_OPT_ZT_CONN_TRIES,
	NNG_OPT_ZT_ORBIT,
	NNG_OPT_ZT_DEORBIT
};

}

inline const char* to_name( option o ) noexcept {
	return detail::option_names[(int)o];
}

inline size_t get_opt_mtu( nng::pipe_view s ) {
	return s.get_opt_size( to_name(option::mtu) );
}

inline uint64_t get_opt_nwid( nng::pipe_view s ) {
	return s.get_opt_uint64( to_name(option::nwid) );
}

inline uint64_t get_opt_node( nng::pipe_view s ) {
	return s.get_opt_uint64( to_name(option::node) );
}

inline buffer get_opt_home( nng::dialer_view s ) {
	return s.get_opt_string( to_name(option::home) );
}

inline void set_opt_home( nng::dialer_view s, const char* v ) {
	s.set_opt_string( to_name(option::home), v );
}

inline uint64_t get_opt_node( nng::dialer_view s ) {
	return s.get_opt_uint64( to_name(option::node) );
}

inline uint64_t get_opt_nwid( nng::dialer_view s ) {
	return s.get_opt_uint64( to_name(option::nwid) );
}

inline status get_opt_network_status( nng::dialer_view s ) {
	return (status)s.get_opt_int( to_name(option::network_status) );
}

inline buffer get_opt_network_name( nng::dialer_view s ) {
	return s.get_opt_string( to_name(option::network_name) );
}

inline nng_duration get_opt_ping_time( nng::dialer_view s ) {
	return s.get_opt_ms( to_name(option::ping_time) );
}

inline void set_opt_ping_time( nng::dialer_view s, nng_duration v ) {
	s.set_opt_ms( to_name(option::ping_time), v );
}

inline int get_opt_ping_tries( nng::dialer_view s ) {
	return s.get_opt_int( to_name(option::ping_tries) );
}

inline void set_opt_ping_tries( nng::dialer_view s, int v ) {
	s.set_opt_int( to_name(option::ping_tries), v );
}

inline nng_duration get_opt_conn_time( nng::dialer_view s ) {
	return s.get_opt_ms( to_name(option::conn_time) );
}

inline void set_opt_conn_time( dialer_view s, nng_duration v ) {
	s.set_opt_ms( to_name(option::conn_time), v );
}

inline int get_opt_conn_tries( dialer_view s ) {
	return s.get_opt_int( to_name(option::conn_tries) );
}

inline void set_opt_conn_tries( dialer_view s, int v ) {
	s.set_opt_int( to_name(option::conn_tries), v );
}

inline uint64_t get_opt_orbit( dialer_view s ) {
	return s.get_opt_uint64( to_name(option::orbit) );
}

inline void set_opt_orbit( dialer_view s, uint64_t v ) {
	s.set_opt_uint64( to_name(option::orbit), v );
}

inline uint64_t get_opt_deorbit( dialer_view s ) {
	return s.get_opt_uint64( to_name(option::deorbit) );
}

inline void set_opt_deorbit( dialer_view s, uint64_t v ) {
	s.set_opt_uint64( to_name(option::deorbit), v );
}

inline buffer get_opt_home( listener_view s ) {
	return s.get_opt_string( to_name(option::home) );
}

inline void set_opt_home( listener_view s, const char* v ) {
	s.set_opt_string( to_name(option::home), v );
}

inline uint64_t get_opt_node( listener_view s ) {
	return s.get_opt_uint64( to_name(option::node) );
}

inline uint64_t get_opt_nwid( listener_view s ) {
	return s.get_opt_uint64( to_name(option::nwid) );
}

inline status get_opt_network_status( listener_view s ) {
	return (status)s.get_opt_int( to_name(option::network_status) );
}

inline buffer get_opt_network_name( listener_view s ) {
	return s.get_opt_string( to_name(option::network_name) );
}

inline nng_duration get_opt_ping_time( listener_view s ) {
	return s.get_opt_ms( to_name(option::ping_time) );
}

inline void set_opt_ping_time( listener_view s, nng_duration v ) {
	s.set_opt_ms( to_name(option::ping_time), v );
}

inline int get_opt_ping_tries( listener_view s ) {
	return s.get_opt_int( to_name(option::ping_tries) );
}

inline void set_opt_ping_tries( listener_view s, int v ) {
	s.set_opt_int( to_name(option::ping_tries), v );
}

inline uint64_t get_opt_orbit( listener_view s ) {
	return s.get_opt_uint64( to_name(option::orbit) );
}

inline void set_opt_orbit( listener_view s, uint64_t v ) {
	s.set_opt_uint64( to_name(option::orbit), v );
}

inline uint64_t get_opt_deorbit( listener_view s ) {
	return s.get_opt_uint64( to_name(option::deorbit) );
}

inline void set_opt_deorbit( listener_view s, uint64_t v ) {
	s.set_opt_uint64( to_name(option::deorbit), v );
}

}}

#endif


<|start_filename|>include/nngpp/msg_header.h<|end_filename|>
#ifndef NNGPP_MSG_HEADER_H
#define NNGPP_MSG_HEADER_H
#include "view.h"

namespace nng {

class msg_header {
	nng_msg* m;
	
public:
	
	explicit msg_header( nng_msg* m ) noexcept : m(m) {}
	
	nng_msg* get_msg() const noexcept {
		return m;
	}

	template<typename T = void>
	T* data() const noexcept {
		return (T*)nng_msg_header(m);
	}

	size_t size() const noexcept {
		return nng_msg_header_len(m);
	}

	view get() const noexcept {
		return view(
			nng_msg_header(m),
			nng_msg_header_len(m)
		);
	}

	void clear() const noexcept {
		nng_msg_header_clear(m);
	}
	
	void append( view v ) const {
		int r = nng_msg_header_append(m,v.data(),v.size());
		if( r != 0 ) {
			throw exception(r,"nng_msg_header_append");
		}
	}
	void insert( view v ) const {
		int r = nng_msg_header_insert(m,v.data(),v.size());
		if( r != 0 ) {
			throw exception(r,"nng_msg_header_insert");
		}
	}
	
	void trim( size_t size ) const {
		int r = nng_msg_header_trim(m,size);
		if( r != 0 ) {
			throw exception(r,"nng_msg_header_trim");
		}
	}
	
	void chop( size_t size ) const {
		int r = nng_msg_header_chop(m,size);
		if( r != 0 ) {
			throw exception(r,"nng_msg_header_chop");
		}
	}

	void append_u16( uint16_t val ) const {
		int r = nng_msg_header_append_u16(m,val);
		if( r != 0 ) {
			throw exception(r,"nng_msg_header_append_u16");
		}
	}

	void append_u32( uint32_t val ) const {
		int r = nng_msg_header_append_u32(m,val);
		if( r != 0 ) {
			throw exception(r,"nng_msg_header_append_u32");
		}
	}
	
	void append_u64( uint64_t val ) const {
		int r = nng_msg_header_append_u64(m,val);
		if( r != 0 ) {
			throw exception(r,"nng_msg_header_append_u64");
		}
	}
	
	void insert_u16( uint16_t val ) const {
		int r = nng_msg_header_insert_u16(m,val);
		if( r != 0 ) {
			throw exception(r,"nng_msg_header_insert_u16");
		}
	}
	
	void insert_u32( uint32_t val ) const {
		int r = nng_msg_header_insert_u32(m,val);
		if( r != 0 ) {
			throw exception(r,"nng_msg_header_insert_u32");
		}
	}
	
	void insert_u64( uint64_t val ) const {
		int r = nng_msg_header_insert_u64(m,val);
		if( r != 0 ) {
			throw exception(r,"nng_msg_header_insert_u64");
		}
	}
	
	uint16_t trim_u16() const {
		uint16_t val;
		int r = nng_msg_header_trim_u16(m,&val);
		if( r != 0 ) {
			throw exception(r,"nng_msg_header_trim_u16");
		}
		return val;
	}
	
	uint32_t trim_u32() const {
		uint32_t val;
		int r = nng_msg_header_trim_u32(m,&val);
		if( r != 0 ) {
			throw exception(r,"nng_msg_header_trim_u32");
		}
		return val;
	}
	
	uint64_t trim_u64() const {
		uint64_t val;
		int r = nng_msg_header_trim_u64(m,&val);
		if( r != 0 ) {
			throw exception(r,"nng_msg_header_trim_u64");
		}
		return val;
	}
	
	uint16_t chop_u16() const {
		uint16_t val;
		int r = nng_msg_header_chop_u16(m,&val);
		if( r != 0 ) {
			throw exception(r,"nng_msg_header_chop_u16");
		}
		return val;
	}
	
	uint32_t chop_u32() const {
		uint32_t val;
		int r = nng_msg_header_chop_u32(m,&val);
		if( r != 0 ) {
			throw exception(r,"nng_msg_header_chop_u32");
		}
		return val;
	}
	
	uint64_t chop_u64() const {
		uint64_t val;
		int r = nng_msg_header_chop_u64(m,&val);
		if( r != 0 ) {
			throw exception(r,"nng_msg_header_chop_u64");
		}
		return val;
	}

};

}

#endif


<|start_filename|>include/nngpp/http/conn_view.h<|end_filename|>
#ifndef NNGPP_HTTP_CONN_VIEW_H
#define NNGPP_HTTP_CONN_VIEW_H
#include <nngpp/aio_view.h>
#include "req_view.h"
#include "res_view.h"

namespace nng { namespace http {

struct conn_view {
protected:
	nng_http_conn* n = nullptr;
	
public:
	conn_view() = default;

	conn_view( nng_http_conn* n ) noexcept : n(n) {}
	
	nng_http_conn* get() const noexcept {
		return n;
	}

	nng_http_conn* operator->() const noexcept {
		return n;
	}

	explicit operator bool() const noexcept {
		return n != nullptr;
	}
	
	void read( aio_view a ) const noexcept {
		nng_http_conn_read(n,a.get());
	}
	
	void read_all( aio_view a ) const noexcept {
		nng_http_conn_read_all(n,a.get());
	}
	
	void read( req_view q, aio_view a ) const noexcept {
		nng_http_conn_read_req(n,q.get(),a.get());
	}
	
	void read( res_view s, aio_view a ) const noexcept {
		nng_http_conn_read_res(n,s.get(),a.get());
	}
	
	void write( aio_view a ) const noexcept {
		nng_http_conn_write(n,a.get());
	}
	
	void write_all( aio_view a ) const noexcept {
		nng_http_conn_write_all(n,a.get());
	}
	
	void write( req_view q, aio_view a ) const noexcept {
		nng_http_conn_write_req(n,q.get(),a.get());
	}
	
	void write( res_view s, aio_view a ) const noexcept {
		nng_http_conn_write_res(n,s.get(),a.get());
	}
	
	void hijack() const {
		int r = nng_http_hijack(n);
		if( r != 0 ) {
			throw exception(r,"nng_http_hijack");
		}
	}

	void transact( req_view q, res_view s, aio_view a ) const noexcept {
		nng_http_conn_transact(n,q.get(),s.get(),a.get());
	}
};

}}

#endif


<|start_filename|>include/nngpp/platform/cv.h<|end_filename|>
#ifndef NNGPP_CV_H
#define NNGPP_CV_H
#include "cv_view.h"
#include "mtx_view.h"

namespace nng {

struct cv : cv_view {

	cv() = default;
	
	explicit cv( nng_cv* c ) noexcept : cv_view(c) {}
	
	explicit cv( mtx_view mut ) {
		int r = nng_cv_alloc( &c, mut.get() );
		if( r != 0 ) {
			throw exception(r,"nng_cv_alloc");
		}
	}
	
	cv( const cv& rhs ) = delete;
	
	cv( cv&& rhs ) noexcept : cv_view(rhs.c) {
		rhs.c = nullptr;
	}
	
	cv& operator=( const cv& rhs ) = delete;
	
	cv& operator=( cv&& rhs ) noexcept {
		if( this != &rhs ) {
			if( c != nullptr ) nng_cv_free(c);
			c = rhs.c;
			rhs.c = nullptr;
		}
		return *this;
	}
	
	~cv() {
		if( c != nullptr ) nng_cv_free(c);
	}
	
	nng_cv* release() noexcept {
		auto out = c;
		c = nullptr;
		return out;
	}

};

inline cv make_cv( mtx_view mut ) {
	return cv(mut);
}

}

#endif


<|start_filename|>test/bus.cpp<|end_filename|>
#include <catch2/catch.hpp>
#include <nngpp/nngpp.h>
#include <nngpp/protocol/bus0.h>

TEST_CASE("BUS pattern","[bus]") {
	auto addr = "inproc://bus";

	SECTION("We can create a BUS socket") {
		auto bus = nng::bus::open();
		REQUIRE(bus);
	}

	SECTION("We can create a linked BUS topology") {

		auto bus1 = nng::bus::open();
		REQUIRE(bus1);
		auto bus2 = nng::bus::open();
		REQUIRE(bus2);
		auto bus3 = nng::bus::open();
		REQUIRE(bus3);

		REQUIRE_NOTHROW( bus1.listen(addr) );
		REQUIRE_NOTHROW( bus2.dial(addr) );
		REQUIRE_NOTHROW( bus3.dial(addr) );

		nng_duration rtimeo = 50;
		REQUIRE_NOTHROW( nng::set_opt_recv_timeout( bus1, rtimeo ) );
		REQUIRE_NOTHROW( nng::set_opt_recv_timeout( bus2, rtimeo ) );
		REQUIRE_NOTHROW( nng::set_opt_recv_timeout( bus3, rtimeo ) );

		SECTION("Messages delivered") {
			auto expected_error = nng::exception(nng::error::timedout).what();

			REQUIRE_THROWS_WITH( bus1.recv_msg(), expected_error );
			REQUIRE_THROWS_WITH( bus2.recv_msg(), expected_error );
			REQUIRE_THROWS_WITH( bus3.recv_msg(), expected_error );
			
			{
				auto msg = nng::make_msg(0);
				REQUIRE(msg);

				msg.body().append("99bits");
				REQUIRE_NOTHROW( bus2.send( std::move(msg) ) );
				REQUIRE_NOTHROW( msg = bus1.recv_msg() );
				REQUIRE( msg.body().get() == "99bits" );

				REQUIRE_THROWS_WITH( msg = bus3.recv_msg(), expected_error );
			}
			{
				auto msg = nng::make_msg(0);
				REQUIRE(msg);

				msg.body().append("onthe");
				REQUIRE_NOTHROW( bus1.send( std::move(msg) ) );

				REQUIRE_NOTHROW( msg = bus2.recv_msg() );
				REQUIRE( msg.body().get() == "onthe" );
				REQUIRE_NOTHROW( msg = bus3.recv_msg() );
				REQUIRE( msg.body().get() == "onthe" );
			}
		}
	}
}


<|start_filename|>include/nngpp/aio_view.h<|end_filename|>
#ifndef NNGPP_AIO_VIEW_H
#define NNGPP_AIO_VIEW_H
#include "msg.h"

namespace nng {

struct aio_view {
protected:
	nng_aio* a = nullptr;
	
public:
	aio_view() = default;

	aio_view( nng_aio* a ) noexcept : a(a) {}

	nng_aio* get() const noexcept {
		return a;
	}

	nng_aio* operator->() const noexcept {
		return a;
	}

	explicit operator bool() const noexcept {
		return a != nullptr;
	}
	
	void stop() const noexcept {
		nng_aio_stop(a);
	}
	
	error result() const noexcept {
		return (error)nng_aio_result(a);
	}
	
	size_t count() const noexcept {
		return nng_aio_count(a);
	}
	
	void cancel() const noexcept {
		nng_aio_cancel(a);
	}
	
	void abort( int err_code ) const noexcept {
		nng_aio_abort(a,err_code);
	}
	
	void wait() const noexcept {
		nng_aio_wait(a);
	}
	
	void set_msg( msg_view m ) const noexcept {
		nng_aio_set_msg(a,m.get());
	}

	void set_msg( msg&& m ) const noexcept {
		set_msg(m.release());
	}
	
	msg_view get_msg() const noexcept {
		return nng_aio_get_msg(a);
	}
	
	msg release_msg() const noexcept {
		auto m = nng_aio_get_msg(a);
		nng_aio_set_msg(a,nullptr);
		return msg(m);
	}

	template<typename T = void>
	void set_input( unsigned int index, T* arg ) const {
		int r = nng_aio_set_input(a,index,(void*)arg);
		if( r != 0 ) {
			throw exception(r,"nng_aio_set_input");
		}
	}

	template<typename T = void>
	T* get_input( unsigned int index ) const noexcept {
		return (T*)nng_aio_get_input(a,index);
	}
	
	template<typename T = void>
	void set_output( unsigned int index, T* arg ) const {
		int r = nng_aio_set_output(a,index,(void*)arg);
		if( r != 0 ) {
			throw exception(r,"nng_aio_set_output");
		}
	}
	
	template<typename T = void>
	T* get_output( unsigned int index ) const noexcept {
		return (T*)nng_aio_get_output(a,index);
	}
	
	void set_timeout( nng_duration when ) const noexcept {
		nng_aio_set_timeout(a,when);
	}
	
	void set_iov( unsigned int niov, const nng_iov* iov ) const {
		int r = nng_aio_set_iov(a,niov,iov);
		if( r != 0 ) {
			throw exception(r,"nng_aio_set_iov");
		}
	}
	
	void set_iov( const nng_iov& iov ) const {
		set_iov(1,&iov);
	}
	
	bool begin() const noexcept {
		return nng_aio_begin(a);
	}
	
	void finish( error e = error::success ) const noexcept {
		nng_aio_finish(a,(int)e);
	}
	
	void defer( nng_aio_cancelfn fn, void* arg ) const noexcept {
		nng_aio_defer(a,fn,arg);
	}
	
};

inline void sleep( nng_duration ms, aio_view a ) noexcept {
	nng_sleep_aio(ms,a.get());
}

}

#endif


<|start_filename|>include/nngpp/transport/tls/config.h<|end_filename|>
#ifndef NNGPP_TLS_CONFIG_H
#define NNGPP_TLS_CONFIG_H
#include "config_view.h"

namespace nng { namespace tls {

struct config : config_view {

	config() = default;
	
	explicit config( nng_tls_config* c ) noexcept : config_view(c) {}
	
	explicit config( mode mode ) {
		int r = nng_tls_config_alloc( &c, (nng_tls_mode)mode );
		if( r != 0 ) {
			throw exception(r,"nng_tls_config_alloc");
		}
	}
	
	config( const config& rhs ) = delete;
	
	config( config&& rhs ) noexcept : config_view(rhs.c) {
		rhs.c = nullptr;
	}
	
	config& operator=( const config& rhs ) = delete;
	
	config& operator=( config&& rhs ) noexcept {
		if( this != &rhs ) {
			if( c != nullptr ) nng_tls_config_free(c);
			c = rhs.c;
			rhs.c = nullptr;
		}
		return *this;
	}
	
	~config() {
		if( c != nullptr ) nng_tls_config_free(c);
	}
	
	nng_tls_config* release() noexcept {
		auto out = c;
		c = nullptr;
		return out;
	}

};

inline config config_view::hold() const noexcept {
	nng_tls_config_hold(c);
	return config(c);
}

inline config make_config( mode mode ) {
	return config(mode);
}

}}

#endif


<|start_filename|>perf/latency.h<|end_filename|>
#ifndef NNGPP_PERF_LATENCY_H
#define NNGPP_PERF_LATENCY_H
#include <nngpp/nngpp.h>
#include <nngpp/platform/platform.h>
#include <nngpp/protocol/pair1.h>
#include "misc.h"

void latency_client(const char* addr, size_t msgsize, int trips) {
	nng_time start, end;
	{
		auto s = nng::pair::open();
		s.dial(addr);
		nng::msg msg(msgsize);

		start = nng::clock();
		for(int i = 0; i < trips; ++i) {
			s.send( std::move(msg) );
			msg = s.recv_msg();
		}
		end = nng::clock();
	}

	auto total = (float)(end - start) / 1000;
	auto latency = (float)(total * 1000000) / (trips * 2);
	printf("total time: %.3f [s]\n", total);
	printf("message size: %d [B]\n", (int)msgsize);
	printf("round trip count: %d\n", trips);
	printf("average latency: %.3f [us]\n", latency);
}

void latency_server(const char* addr, size_t msgsize, int trips) {
	auto s = nng::pair::open();
	s.listen(addr);
	for(int i = 0; i < trips; ++i) {
		auto msg = s.recv_msg();
		if( msg.body().size() != msgsize ) {
			die("wrong message size: %d != %d", msg.body().size(), msgsize);
		}
		s.send( std::move(msg) );
	}

	// Wait a bit for things to drain... linger should do this.
	// 100ms ought to be enough.
	nng::msleep(100);
}

#endif


<|start_filename|>include/nngpp/url_view.h<|end_filename|>
#ifndef NNGPP_URL_VIEW_H
#define NNGPP_URL_VIEW_H
#include "view.h"

namespace nng {

struct url_view {
protected:
	nng_url* u = nullptr;
	
public:
	url_view() = default;
	
	url_view( nng_url* u ) noexcept : u(u) {}
	
	nng_url* get() const noexcept {
		return u;
	}
	
	nng_url* operator->() const noexcept {
		return u;
	}

	explicit operator bool() const noexcept {
		return u != nullptr;
	}

	bool operator==( const url_view& rhs ) const noexcept {
		view view_left(u->u_rawurl,strlen(u->u_rawurl));
		view view_right(rhs.u->u_rawurl,strlen(rhs.u->u_rawurl));
		return view_left == view_right;
	}

	bool operator!=( const url_view& rhs ) const noexcept {
		return !(*this == rhs);
	}

};

}

#endif


<|start_filename|>include/nngpp/protocol/pair0.h<|end_filename|>
#ifndef NNGPP_PAIR0_H
#define NNGPP_PAIR0_H
#include <nngpp/socket.h>
#include <nng/protocol/pair0/pair.h>

namespace nng { namespace pair { namespace v0 {

inline socket open() {
	nng_socket s;
	int r = nng_pair0_open(&s);
	if( r != 0 ) {
		throw exception(r,"nng_pair0_open");
	}
	return socket(s);
}

inline socket open_raw() {
	nng_socket s;
	int r = nng_pair0_open_raw(&s);
	if( r != 0 ) {
		throw exception(r,"nng_pair0_open_raw");
	}
	return socket(s);
}

}}}

#endif


<|start_filename|>include/nngpp/stream/dialer.h<|end_filename|>
#ifndef NNGPP_STREAM_DIALER_H
#define NNGPP_STREAM_DIALER_H
#include "dialer_view.h"
#include <nngpp/url_view.h>

namespace nng { namespace stream {

struct dialer : dialer_view {

	dialer() = default;
	
	explicit dialer( nng_stream_dialer* d ) noexcept : dialer_view(d) {}

	explicit dialer( const char* url ) {
		int r = nng_stream_dialer_alloc( &d, url );
		if( r != 0 ) {
			throw exception(r,"nng_stream_dialer_alloc");
		}
	}

	explicit dialer( nng::url_view url ) {
		int r = nng_stream_dialer_alloc_url( &d, url.get() );
		if( r != 0 ) {
			throw exception(r,"nng_stream_dialer_alloc_url");
		}
	}

	dialer( const dialer& rhs ) = delete;
	
	dialer( dialer&& rhs ) noexcept : dialer_view(rhs.d) {
		rhs.d = nullptr;
	}
	
	dialer& operator=( const dialer& rhs ) = delete;
	
	dialer& operator=( dialer&& rhs ) noexcept {
		if( this != &rhs ) {
			if( d != nullptr ) nng_stream_dialer_free(d);
			d = rhs.d;
			rhs.d = nullptr;
		}
		return *this;
	}
	
	~dialer() {
		if( d != nullptr ) nng_stream_dialer_free(d);
	}
	
	nng_stream_dialer* release() noexcept {
		auto out = d;
		d = nullptr;
		return out;
	}

};

inline dialer make_dialer( const char* url ) {
	return dialer(url);
}

inline dialer make_dialer( nng::url_view url ) {
	return dialer(url);
}

}}

#endif


<|start_filename|>include/nngpp/protocol/req0.h<|end_filename|>
#ifndef NNGPP_REQ0_H
#define NNGPP_REQ0_H
#include <nngpp/ctx.h>
#include <nng/protocol/reqrep0/req.h>

namespace nng { namespace req {
inline namespace v0 {

inline socket open() {
	nng_socket s;
	int r = nng_req0_open(&s);
	if( r != 0 ) {
		throw exception(r,"nng_req0_open");
	}
	return socket(s);
}

inline socket open_raw() {
	nng_socket s;
	int r = nng_req0_open_raw(&s);
	if( r != 0 ) {
		throw exception(r,"nng_req0_open_raw");
	}
	return socket(s);
}

enum class option {
	resend_time // ms
};

namespace detail {

#if __cpp_inline_variables >= 201606
inline
#else
static
#endif
const char* const option_names[] = {
	NNG_OPT_REQ_RESENDTIME
};

}

inline const char* to_name( option o ) noexcept {
	return detail::option_names[(int)o];
}

inline nng_duration get_opt_resend_time( socket_view s ) {
	return s.get_opt_ms( to_name(option::resend_time) );
}

inline void set_opt_resend_time( socket_view s, nng_duration v ) {
	s.set_opt_ms( to_name(option::resend_time), v );
}

inline nng_duration get_opt_resend_time( ctx_view s ) {
	return s.get_opt_ms( to_name(option::resend_time) );
}

inline void set_opt_resend_time( ctx_view s, nng_duration v ) {
	s.set_opt_ms( to_name(option::resend_time), v );
}

}
}}

#endif


<|start_filename|>test/ipcsupp.cpp<|end_filename|>
#include <catch2/catch.hpp>
#include <cstring>
#include <nngpp/nngpp.h>

TEST_CASE("Supplemental IPC", "[ipc]") {
	
	INFO("We can create a dialer and listener");
	auto url = "ipc:///tmp/ipcsupp_test";

	auto d = nng::stream::dialer(url);
	REQUIRE(d);
	auto l = nng::stream::listener(url);
	REQUIRE(l);
	
	INFO("Listener listens");
	REQUIRE_NOTHROW(l.listen());

	INFO("We can dial it");
	nng::aio daio(nullptr,nullptr);
	nng::aio laio(nullptr,nullptr);
	nng::aio maio(nullptr,nullptr);

	d.dial(daio);
	l.accept(laio);

	daio.wait();
	laio.wait();

	REQUIRE(daio.result() == nng::error::success);
	REQUIRE(laio.result() == nng::error::success);

	nng::stream::stream c1( daio.get_output<nng_stream>(0) );
	nng::stream::stream c2( laio.get_output<nng_stream>(0) );
	REQUIRE(c1);
	REQUIRE(c2);
	
	INFO("They exchange messages");
	nng::aio aio1(nullptr,nullptr);
	nng::aio aio2(nullptr,nullptr);
	
	char buf1[5];
	memcpy(buf1, "TEST", 5);

	char buf2[5];
	memset(buf2, 0, 5);
	
	aio1.set_iov( {buf1, 5} );
	aio2.set_iov( {buf2, 5} );
	c1.send(aio1);
	c2.recv(aio2);
	aio1.wait();
	aio2.wait();

	REQUIRE(aio1.result() == nng::error::success);
	REQUIRE(aio1.count() == 5);

	REQUIRE(aio2.result() == nng::error::success);
	REQUIRE(aio2.count() == 5);

	REQUIRE(memcmp(buf1, buf2, 5) == 0);

	INFO("Socket name matches");
	{
		nng_sockaddr sa2;
		REQUIRE_NOTHROW(sa2 = c2.get_addr(to_name(nng::option::local_address)));
		REQUIRE(sa2.s_ipc.sa_family == NNG_AF_IPC);
		REQUIRE(strcmp(sa2.s_ipc.sa_path, url + strlen("ipc://")) == 0);
	}

}


<|start_filename|>include/nngpp/url.h<|end_filename|>
#ifndef NNGPP_URL_H
#define NNGPP_URL_H
#include "url_view.h"

namespace nng {

struct url : url_view {

	url() = default;

	explicit url( nng_url* u ) noexcept : url_view(u) {}
	
	explicit url( const char* str ) {
		int r = nng_url_parse( &u, str );
		if( r != 0 ) {
			throw exception(r,"nng_url_parse");
		}
	}
	
	url( const url& rhs ) : url_view() {
		int r = nng_url_clone(&u,rhs.get());
		if( r != 0 ) {
			throw exception(r,"nng_url_parse");
		}
	}
	
	url( url&& rhs ) noexcept : url_view(rhs.u) {
		rhs.u = nullptr;
	}
	
	url& operator=( const url& rhs ) {
		if( this != &rhs ) {
			if( u != nullptr ) nng_url_free(u);
			int r = nng_url_clone(&u,rhs.get());
			if( r != 0 ) {
				throw exception(r,"nng_url_clone");
			}
		}
		return *this;
	}
	
	url& operator=( url&& rhs ) noexcept {
		if( this != &rhs ) {
			if( u != nullptr ) nng_url_free(u);
			u = rhs.u;
			rhs.u = nullptr;
		}
		return *this;
	}
	
	~url() {
		if( u != nullptr ) nng_url_free(u);
	}
	
	nng_url* release() noexcept {
		auto out = u;
		u = nullptr;
		return out;
	}

};

inline url make_url( const char* str ) {
	return url(str);
}

}

#endif


<|start_filename|>include/nngpp/stat.h<|end_filename|>
#ifndef NNGPP_STAT_H
#define NNGPP_STAT_H
#include "stat_view.h"
#include "error.h"

namespace nng {

struct stat : stat_view {

	stat() = default;

	explicit stat( nng_stat* s ) noexcept : stat_view(s) {}

	stat( const stat& rhs ) = delete;
	
	stat( stat&& rhs ) noexcept : stat_view(rhs.s) {
		rhs.s = nullptr;
	}
	
	stat& operator=( const stat& rhs ) = delete;
	
	stat& operator=( stat&& rhs ) noexcept {
		if( this != &rhs ) {
			if( s != nullptr ) nng_stats_free(s);
			s = rhs.s;
			rhs.s = nullptr;
		}
		return *this;
	}
	
	~stat() {
		if( s != nullptr ) nng_stats_free(s);
	}

	nng_stat* release() noexcept {
		auto out = s;
		s = nullptr;
		return out;
	}

};

inline stat make_stat() {
	nng_stat* s;
	int r = nng_stats_get(&s);
	if( r != 0 ) {
		throw exception(r,"nng_stats_get");
	}
	return stat(s);
}

}

#endif


<|start_filename|>include/nngpp/http/req.h<|end_filename|>
#ifndef NNGPP_HTTP_REQ_H
#define NNGPP_HTTP_REQ_H
#include "req_view.h"
#include <nngpp/url_view.h>

namespace nng { namespace http {

struct req : req_view {

	req() = default;

	explicit req( nng_http_req* q ) noexcept : req_view(q) {}
	
	explicit req( url_view u ) {
		int r = nng_http_req_alloc( &q, u.get() );
		if( r != 0 ) {
			throw exception(r,"nng_http_req_alloc");
		}
	}
	
	req( const req& rhs ) = delete;
	
	req( req&& rhs ) noexcept : req_view(rhs.q) {
		rhs.q = nullptr;
	}
	
	req& operator=( const req& rhs ) = delete;
	
	req& operator=( req&& rhs ) noexcept {
		if( this != &rhs ) {
			if( q != nullptr ) nng_http_req_free(q);
			q = rhs.q;
			rhs.q = nullptr;
		}
		return *this;
	}
	
	~req() {
		if( q != nullptr ) nng_http_req_free(q);
	}

	nng_http_req* release() noexcept {
		auto out = q;
		q = nullptr;
		return out;
	}

};

inline req make_req( url_view u ) {
	return req(u);
}

}}

#endif


<|start_filename|>include/nngpp/pipe_view.h<|end_filename|>
#ifndef NNGPP_PIPE_VIEW_H
#define NNGPP_PIPE_VIEW_H
#include "dialer_view.h"
#include "listener_view.h"

namespace nng {

struct socket_view;

struct pipe_view {
protected:
	nng_pipe p{0};
	
public:
	pipe_view() = default;
	
	pipe_view( nng_pipe p ) noexcept : p(p) {}
	
	nng_pipe get() const noexcept {
		return p;
	}

	explicit operator bool() const noexcept {
		return p.id != 0;
	}
	
	int id() const noexcept {
		return nng_pipe_id(p);
	}
	
	size_t get_opt( const char* name, buffer& out ) const {
		size_t size = out.size();
		int r = nng_pipe_get(p,name,out.data(),&size);
		if( r != 0 ) {
			throw exception(r,"nng_pipe_get");
		}
		return size;
	}

	bool get_opt_bool( const char* name ) const {
		bool out;
		int r = nng_pipe_get_bool(p,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_pipe_get_bool");
		}
		return out;
	}
	
	int get_opt_int( const char* name ) const {
		int out;
		int r = nng_pipe_get_int(p,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_pipe_get_int");
		}
		return out;
	}
	
	size_t get_opt_size( const char* name ) const {
		size_t out;
		int r = nng_pipe_get_size(p,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_pipe_get_size");
		}
		return out;
	}

	uint64_t get_opt_uint64( const char* name ) const {
		uint64_t out;
		int r = nng_pipe_get_uint64(p,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_pipe_get_uint64");
		}
		return out;
	}
	
	buffer get_opt_string( const char* name ) const {
		char* data;
		int r = nng_pipe_get_string(p,name,&data);
		if( r != 0 ) {
			throw exception(r,"nng_pipe_get_string");
		}
		return buffer(data,strlen(data)+1);
	}
	
	void* get_opt_ptr( const char* name ) const {
		void* out;
		int r = nng_pipe_get_ptr(p,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_pipe_get_ptr");
		}
		return out;
	}
	
	nng_duration get_opt_ms( const char* name ) const {
		nng_duration out;
		int r = nng_pipe_get_ms(p,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_pipe_get_ms");
		}
		return out;
	}

	nng_sockaddr get_opt_addr( const char* name ) const {
		nng_sockaddr out;
		int r = nng_pipe_get_addr(p,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_pipe_get_addr");
		}
		return out;
	}

	socket_view get_socket() const noexcept;

	dialer_view get_dialer() const noexcept {
		return nng_pipe_dialer(p);
	}
	
	listener_view get_listener() const noexcept {
		return nng_pipe_listener(p);
	}
	
};

inline nng_sockaddr get_opt_local_address( pipe_view s ) {
	return s.get_opt_addr( to_name(option::local_address) );
}

inline nng_sockaddr get_opt_remote_address( pipe_view s ) {
	return s.get_opt_addr( to_name(option::remote_address) );
}

}

#endif


<|start_filename|>include/nngpp/http/conn.h<|end_filename|>
#ifndef NNGPP_HTTP_CONN_H
#define NNGPP_HTTP_CONN_H
#include "conn_view.h"

namespace nng { namespace http {

struct conn : conn_view {

	conn() = default;

	explicit conn( nng_http_conn* n ) noexcept : conn_view(n) {}
	
	conn( const conn& rhs ) = delete;
	
	conn( conn&& rhs ) noexcept : conn_view(rhs.n) {
		rhs.n = nullptr;
	}
	
	conn& operator=( const conn& rhs ) = delete;
	
	conn& operator=( conn&& rhs ) noexcept {
		if( this != &rhs ) {
			if( n != nullptr ) nng_http_conn_close(n);
			n = rhs.n;
			rhs.n = nullptr;
		}
		return *this;
	}
	
	~conn() {
		if( n != nullptr ) nng_http_conn_close(n);
	}
	
	nng_http_conn* release() noexcept {
		auto out = n;
		n = nullptr;
		return out;
	}

};

}}

#endif


<|start_filename|>include/nngpp/protocol/push0.h<|end_filename|>
#ifndef NNGPP_PUSHE0_H
#define NNGPP_PUSHE0_H
#include <nngpp/socket.h>
#include <nng/protocol/pipeline0/push.h>

namespace nng { namespace push {
inline namespace v0 {

inline socket open() {
	nng_socket s;
	int r = nng_push0_open(&s);
	if( r != 0 ) {
		throw exception(r,"nng_push0_open");
	}
	return socket(s);
}

inline socket open_raw() {
	nng_socket s;
	int r = nng_push0_open_raw(&s);
	if( r != 0 ) {
		throw exception(r,"nng_push0_open_raw");
	}
	return socket(s);
}

}
}}

#endif


<|start_filename|>demo/http_client/http_client.cpp<|end_filename|>
// This is a port of the nng demo to nngpp
// See https://github.com/nanomsg/nng/tree/master/demo/http_client

#include <cstdio>
#include <cstdlib>
#include <nngpp/nngpp.h>
#include <nngpp/http/http.h>

int main( int argc, char** argv ) try {
	if(argc < 2) {
		fprintf(stderr, "No URL supplied!\n");
		return 1;
	}

	nng::url url( argv[1] );
	nng::http::client client( url );

	auto aio = nng::make_aio();
	client.connect( aio );

	aio.wait();
	{
		auto result = aio.result();
		if( result != nng::error::success ) {
			throw nng::exception(result);
		}
	}

	// Get the connection, at the 0th output.
	nng::http::conn_view conn = aio.get_output<nng_http_conn>(0);

	// Request is already set up with URL, and for GET via HTTP/1.1.
	// The Host: header is already set up too.

	// Send the request, and wait for that to finish.
	nng::http::req req(url);
	conn.write(req,aio);
	aio.wait();
	{
		auto result = aio.result();
		if( result != nng::error::success ) {
			throw nng::exception(result);
		}
	}
	
	// Read a response.
	auto res = nng::http::make_res();
	conn.read(res,aio);
	aio.wait();
	{
		auto result = aio.result();
		if( result != nng::error::success ) {
			throw nng::exception(result);
		}
	}

	if(res.get_status() != nng::http::status::ok) {
		fprintf(stderr, "HTTP Server Responded: %d %s\n", (int)res.get_status(), res.get_reason());
	}

	// This only supports regular transfer encoding (no Chunked-Encoding,
	// and a Content-Length header is required.)
	auto hdr = res.get_header("Content-Length");
	if(!hdr) {
		fprintf(stderr, "Missing Content-Length header.\n");
		return 1;
	}

	auto len = atoi(hdr);
	if(len == 0) {
		return 0;
	}

	// Allocate a buffer to receive the body data.
	nng::buffer data(len);

	// Set up a single iov to point to the buffer.
	// Following never fails with fewer than 5 elements.
	aio.set_iov( data );

	// Now attempt to receive the data.
	conn.read_all(aio);

	// Wait for it to complete.
	aio.wait();
	{
		auto result = aio.result();
		if( result != nng::error::success ) {
			throw nng::exception(result);
		}
	}

	fwrite(data.data(), 1, data.size(), stdout);
}
catch( const nng::exception& e ) {
	fprintf(stderr, "%s: %s\n", e.who(), e.what());
	return 1;
}


<|start_filename|>include/nngpp/http/req_view.h<|end_filename|>
#ifndef NNGPP_HTTP_REQ_VIEW_H
#define NNGPP_HTTP_REQ_VIEW_H
#include <nngpp/view.h>
#include "misc.h"

namespace nng { namespace http {

struct req_view {
protected:
	nng_http_req* q = nullptr;
	
public:
	req_view() = default;

	req_view( nng_http_req* q ) noexcept : q(q) {}

	nng_http_req* get() const noexcept {
		return q;
	}

	nng_http_req* operator->() const noexcept {
		return q;
	}

	explicit operator bool() const noexcept {
		return q != nullptr;
	}
	
	void reset() const noexcept {
		nng_http_req_reset(q);
	}

	const char* get_method() const noexcept {
		return nng_http_req_get_method(q);
	}
	
	void set_method( const char* method ) const {
		int r = nng_http_req_set_method(q,method);
		if( r != 0 ) {
			throw exception(r,"nng_http_req_set_method");
		}
	}
	
	const char* get_version() const noexcept {
		return nng_http_req_get_version(q);
	}
	
	void set_version( const char* version ) const {
		int r = nng_http_req_set_version(q,version);
		if( r != 0 ) {
			throw exception(r,"nng_http_req_set_version");
		}
	}
	
	const char* get_uri() const noexcept {
		return nng_http_req_get_uri(q);
	}
	
	void set_uri( const char* uri ) const {
		int r = nng_http_req_set_uri(q,uri);
		if( r != 0 ) {
			throw exception(r,"nng_http_req_set_uri");
		}
	}
	
	const char* get_header( const char* name ) const noexcept {
		return nng_http_req_get_header(q,name);
	}
	
	void set_header( const char* name, const char* value ) const {
		int r = nng_http_req_set_header(q,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_http_req_set_header");
		}
	}
	
	void add_header( const char* name, const char* value ) const {
		int r = nng_http_req_add_header(q,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_http_req_add_header");
		}
	}
	
	void del_header( const char* name ) const {
		int r = nng_http_req_del_header(q,name);
		if( r != 0 ) {
			throw exception(r,"nng_http_req_del_header");
		}
	}

	view get_data() const noexcept {
		void* data;
		size_t size;
		nng_http_req_get_data(q,&data,&size);
		return view(data,size);
	}
	
	void set_data( view v ) const {
		int r = nng_http_req_set_data(q,v.data(),v.size());
		if( r != 0 ) {
			throw exception(r,"nng_http_req_set_data");
		}
	}
	
	void copy_data( view v ) const {
		int r = nng_http_req_copy_data(q,v.data(),v.size());
		if( r != 0 ) {
			throw exception(r,"nng_http_req_copy_data");
		}
	}

};

}}

#endif


<|start_filename|>include/nngpp/core.h<|end_filename|>
#ifndef NNGPP_CORE_H
#define NNGPP_CORE_H
#include <nng/nng.h>

namespace nng {

enum class sockaddr_family {
	unspec = NNG_AF_UNSPEC,
	inproc = NNG_AF_INPROC,
	ipc    = NNG_AF_IPC,
	inet   = NNG_AF_INET,
	inet6  = NNG_AF_INET6,
	zt     = NNG_AF_ZT
};

enum class pipe_ev {
	add_pre  = NNG_PIPE_EV_ADD_PRE,
	add_post = NNG_PIPE_EV_ADD_POST,
	rem_post = NNG_PIPE_EV_REM_POST,
	num      = NNG_PIPE_EV_NUM
};

namespace flag {
enum {
	alloc    = NNG_FLAG_ALLOC,
	nonblock = NNG_FLAG_NONBLOCK
};
}

inline const char* version() noexcept {
	return nng_version();
}

inline void fini() noexcept {
	nng_fini();
}

inline void close_all() noexcept {
	nng_closeall();
}

}

#endif


<|start_filename|>test/synch.cpp<|end_filename|>
#include <catch2/catch.hpp>
#include <nngpp/nngpp.h>
#include <nngpp/platform/platform.h>
#include <mutex>

namespace {

struct notifyarg {
	int          did = 0;
	nng_duration when = 0;
	nng::mtx     mx = nng::make_mtx();
	nng::cv      cv{mx};
};

static void notifyafter(void* arg) {
	auto na = (notifyarg*)arg;

	nng::msleep(na->when);
	std::lock_guard<nng::mtx> lock( na->mx );
	na->did = 1;
	na->cv.wake_all();
}

}

TEST_CASE("We can lock and unlock a mutex","[synch]") {
	auto mx = nng::make_mtx();
	mx.lock();
	REQUIRE(1);
	mx.unlock();
	REQUIRE(1);
	SECTION("And then lock it again","[synch]") {
		mx.lock();
		REQUIRE(1);
		mx.unlock();
		REQUIRE(1);
	}
}
TEST_CASE("Things block properly","[synch]") {
	notifyarg arg;
	nng::thread thr;
	{
		std::lock_guard<nng::mtx> lock( arg.mx );
		thr = nng::thread(notifyafter,&arg);
		nng::msleep(10);
		REQUIRE(arg.did == 0);
	}
	nng::msleep(10);
	{
		std::lock_guard<nng::mtx> lock( arg.mx );
		while(!arg.did) {
			arg.cv.wait();
		}
		REQUIRE(arg.did != 0);
	}
}

TEST_CASE("Notification works","[synch]") {
	notifyarg arg;
	arg.did  = 0;
	arg.when = 10;
	{
		nng::thread thr(notifyafter,&arg);
		std::lock_guard<nng::mtx> lock( arg.mx );
		if(!arg.did) {
			arg.cv.wait();
		}
	}
	REQUIRE(arg.did == 1);
}

TEST_CASE("Timeout works","[synch]") {
	notifyarg arg;
	arg.did  = 0;
	arg.when = 200;
	nng::thread thr(notifyafter,&arg);
	std::lock_guard<nng::mtx> lock( arg.mx );
	if(!arg.did) {
		arg.cv.wait_until(nng::clock() + 10);
	}
	REQUIRE(arg.did == 0);
}

TEST_CASE("Empty timeout is EAGAIN","[synch]") {
	notifyarg arg;
	std::lock_guard<nng::mtx> lock( arg.mx );
	REQUIRE_THROWS_WITH(
		arg.cv.wait_until(0),
		nng::exception(nng::error::again).what()
	);
}

TEST_CASE("Not running works","[synch]") {
	notifyarg arg;
	arg.did  = 0;
	arg.when = 1;
	std::lock_guard<nng::mtx> lock( arg.mx );
	if(!arg.did) {
		arg.cv.wait_until(nng::clock() + 10);
	}
	REQUIRE(arg.did == 0);
}


<|start_filename|>include/nngpp/socket.h<|end_filename|>
#ifndef NNGPP_SOCKET_H
#define NNGPP_SOCKET_H
#include "socket_view.h"

namespace nng {

struct socket : socket_view {

	socket() = default;

	explicit socket( nng_socket s ) noexcept : socket_view(s) {}
	
	socket( const socket& rhs ) = delete;
	
	socket( socket&& rhs ) noexcept : socket_view(rhs.s) {
		rhs.s.id = 0;
	}
	
	socket& operator=( const socket& rhs ) = delete;
	
	socket& operator=( socket&& rhs ) {
		if( this != &rhs ) {
			if( s.id != 0 ) {
				int r = nng_close(s);
				if( r != 0 && r != (int)nng::error::closed ) {
					throw exception(r,"nng_close");
				}
			}
			s = rhs.s;
			rhs.s.id = 0;
		}
		return *this;
	}
	
	~socket() {
		if( s.id != 0 ) nng_close(s);
	}
	
	nng_socket release() noexcept {
		auto out = s;
		s.id = 0;
		return out;
	}

};

}

#endif


<|start_filename|>include/nngpp/aio.h<|end_filename|>
#ifndef NNGPP_AIO_H
#define NNGPP_AIO_H
#include "aio_view.h"

namespace nng {

struct aio : aio_view {

	aio() = default;

	explicit aio( nng_aio* a ) noexcept : aio_view(a) {}
	
	explicit aio( void (*cb)(void*), void* arg ) {
		int r = nng_aio_alloc( &a, cb, arg );
		if( r != 0 ) {
			throw exception(r,"nng_aio_alloc");
		}
	}
	
	aio( const aio& rhs ) = delete;
	
	aio( aio&& rhs ) noexcept : aio_view(rhs.a) {
		rhs.a = nullptr;
	}
	
	aio& operator=( const aio& rhs ) = delete;
	
	aio& operator=( aio&& rhs ) noexcept {
		if( this != &rhs ) {
			if( a != nullptr ) nng_aio_free(a);
			a = rhs.a;
			rhs.a = nullptr;
		}
		return *this;
	}
	
	~aio() {
		if( a != nullptr ) nng_aio_free(a);
	}

	nng_aio* release() noexcept {
		auto out = a;
		a = nullptr;
		return out;
	}

};

inline aio make_aio( void (*cb)(void*), void* arg ) {
	return aio(cb,arg);
}

inline aio make_aio() {
	return aio(nullptr,nullptr);
}

}

#endif


<|start_filename|>test/scalability.cpp<|end_filename|>
#include <catch2/catch.hpp>
#include <nngpp/nngpp.h>
#include <nngpp/protocol/req0.h>
#include <nngpp/protocol/rep0.h>
#include <nngpp/platform/platform.h>
#include <vector>

namespace {

static const char* addr = "inproc://scalability";

struct server {
	nng::socket rep = nng::rep::open();

	void run() const {
		nng::set_opt_recv_buffer(rep,256);
		nng::set_opt_send_buffer(rep,256);
		rep.listen(addr);

		while(true) {
			try {
				rep.send( rep.recv_msg() );
			}
			catch( const nng::exception& ) {
				break;
			}
		}
	}

	void stop() {
		rep = nng::socket();
	}

};

static void serve(void* arg) {
	((server*)arg)->run();
}

static void open_clients(nng::socket* clients, int num) {
	nng_duration t = 100; // 100ms
	for(int i=0;i<num;++i) {
		CAPTURE(i);
		auto c = nng::req::open();
		REQUIRE_NOTHROW(nng::set_opt_recv_timeout(c,t));
		REQUIRE_NOTHROW(nng::set_opt_send_timeout(c,t));
		nng::msleep(10);
		REQUIRE_NOTHROW(c.dial(addr));
		clients[i] = std::move(c);
	}
}

static void transact(nng::socket* clients, int num) {
	for(int i=0;i<num;++i) {
		CAPTURE(i);
		clients[i].send( nng::make_msg(0) );
		clients[i].recv_msg();
	}
}

}

TEST_CASE("Scalability","[scalability]") {
	int nclients = 200;
	server sv;
	nng::thread server(serve, &sv);
	std::vector<nng::socket> clients(nclients);

	REQUIRE(server);

	REQUIRE_NOTHROW(open_clients(clients.data(), nclients));

	REQUIRE_NOTHROW(transact(clients.data(), nclients));

	sv.stop();
}



<|start_filename|>include/nngpp/http/handler.h<|end_filename|>
#ifndef NNGPP_HTTP_HANDLER_H
#define NNGPP_HTTP_HANDLER_H
#include "handler_view.h"

namespace nng { namespace http {

struct path_type_file {};
struct path_type_dir {};

struct handler : handler_view {

	handler() = default;
	
	explicit handler( nng_http_handler* h ) noexcept : handler_view(h) {}
	
	explicit handler( const char* uri, void (*cb)(nng_aio*) ) {
		int r = nng_http_handler_alloc( &h, uri, cb );
		if( r != 0 ) {
			throw exception(r,"nng_http_handler_alloc");
		}
	}
	
	explicit handler( const char* uri, const char* path, path_type_file ) {
		int r = nng_http_handler_alloc_file( &h, uri, path );
		if( r != 0 ) {
			throw exception(r,"nng_http_handler_alloc_file");
		}
	}
	
	explicit handler( const char* uri, const char* path, path_type_dir ) {
		int r = nng_http_handler_alloc_directory( &h, uri, path );
		if( r != 0 ) {
			throw exception(r,"nng_http_handler_alloc_directory");
		}
	}
	
	explicit handler( const char* uri, const void* data, size_t size, const char* ctype ) {
		int r = nng_http_handler_alloc_static( &h, uri, data, size, ctype );
		if( r != 0 ) {
			throw exception(r,"nng_http_handler_alloc_static");
		}
	}
	
	handler( const handler& rhs ) = delete;
	
	handler( handler&& rhs ) noexcept : handler_view(rhs.h) {
		rhs.h = nullptr;
	}
	
	handler& operator=( const handler& rhs ) = delete;
	
	handler& operator=( handler&& rhs ) noexcept {
		if( this != &rhs ) {
			if( h != nullptr ) nng_http_handler_free(h);
			h = rhs.h;
			rhs.h = nullptr;
		}
		return *this;
	}
	
	~handler() {
		if( h != nullptr ) nng_http_handler_free(h);
	}
	
	nng_http_handler* release() noexcept {
		auto out = h;
		h = nullptr;
		return out;
	}

};

inline handler make_handler( const char* uri, void (*cb)(nng_aio*) ) {
	return handler(uri,cb);
}

inline handler make_handler( const char* uri, const char* path, path_type_file path_type ) {
	return handler(uri,path,path_type);
}

inline handler make_handler( const char* uri, const char* path, path_type_dir path_type ) {
	return handler(uri,path,path_type);
}

inline handler make_handler( const char* uri, const void* data, size_t size, const char* ctype ) {
	return handler(uri,data,size,ctype);
}

}}

#endif


<|start_filename|>include/nngpp/dialer.h<|end_filename|>
#ifndef NNGPP_DIALER_H
#define NNGPP_DIALER_H
#include "dialer_view.h"
#include "socket_view.h"

namespace nng {

struct dialer : dialer_view {

	dialer() = default;
	
	explicit dialer( nng_dialer d ) noexcept : dialer_view(d) {}
	
	explicit dialer( socket_view s, const char* addr, int flags ) {
		int r = nng_dial( s.get(), addr, &d, flags );
		if( r != 0 ) {
			throw exception(r,"nng_dial");
		}
	}
	
	explicit dialer( socket_view s, const char* addr ) {
		int r = nng_dialer_create( &d, s.get(), addr );
		if( r != 0 ) {
			throw exception(r,"nng_dialer_create");
		}
	}
	
	dialer( const dialer& rhs ) = delete;
	
	dialer( dialer&& rhs ) noexcept : dialer_view(rhs.d) {
		rhs.d.id = 0;
	}
	
	dialer& operator=( const dialer& rhs ) = delete;
	
	dialer& operator=( dialer&& rhs ) {
		if( this != &rhs ) {
			if( d.id != 0 ) {
				int r = nng_dialer_close(d);
				if( r != 0 && r != (int)nng::error::closed ) {
					throw exception(r,"nng_dialer_close");
				}
			}
			d = rhs.d;
			rhs.d.id = 0;
		}
		return *this;
	}
	
	~dialer() {
		if( d.id != 0 ) nng_dialer_close(d);
	}
	
	nng_dialer release() noexcept {
		auto out = d;
		d.id = 0;
		return out;
	}

};

inline dialer make_dialer( socket_view s, const char* addr, int flags ) {
	return dialer(s,addr,flags);
}

inline dialer make_dialer( socket_view s, const char* addr ) {
	return dialer(s,addr);
}

}

#endif


<|start_filename|>include/nngpp/protocol/pub0.h<|end_filename|>
#ifndef NNGPP_PUB0_H
#define NNGPP_PUB0_H
#include <nngpp/socket.h>
#include <nng/protocol/pubsub0/pub.h>

namespace nng { namespace pub {
inline namespace v0 {

inline socket open() {
	nng_socket s;
	int r = nng_pub0_open(&s);
	if( r != 0 ) {
		throw exception(r,"nng_pub0_open");
	}
	return socket(s);
}

inline socket open_raw() {
	nng_socket s;
	int r = nng_pub0_open_raw(&s);
	if( r != 0 ) {
		throw exception(r,"nng_pub0_open_raw");
	}
	return socket(s);
}

}
}}

#endif


<|start_filename|>include/nngpp/platform/cv_view.h<|end_filename|>
#ifndef NNGPP_CV_VIEW_H
#define NNGPP_CV_VIEW_H
#include <nngpp/error.h>
#include <nng/supplemental/util/platform.h>

namespace nng {

struct cv_view {
protected:
	nng_cv* c = nullptr;
	
public:
	cv_view() = default;
	
	cv_view( nng_cv* c ) noexcept : c(c) {}

	nng_cv* get() const noexcept {
		return c;
	}

	nng_cv* operator->() const noexcept {
		return c;
	}

	explicit operator bool() const noexcept {
		return c != nullptr;
	}

	void wait() const noexcept {
		nng_cv_wait(c);
	}

	/** wait until time t
	 *  returns true if signalled, false if timed out
	 */
	bool wait_until( nng_time t ) const {
		int r = nng_cv_until(c,t);
		if( r != 0 && r != (int)error::timedout ) {
			throw exception(r,"nng_cv_until");
		}
		return r == 0;
	}

	void wake_all() const noexcept {
		nng_cv_wake(c);
	}

	void wake_one() const noexcept {
		nng_cv_wake1(c);
	}

};

}

#endif


<|start_filename|>include/nngpp/protocol/pull0.h<|end_filename|>
#ifndef NNGPP_PULL0_H
#define NNGPP_PULL0_H
#include <nngpp/socket.h>
#include <nng/protocol/pipeline0/pull.h>

namespace nng { namespace pull {
inline namespace v0 {

inline socket open() {
	nng_socket s;
	int r = nng_pull0_open(&s);
	if( r != 0 ) {
		throw exception(r,"nng_pull0_open");
	}
	return socket(s);
}

inline socket open_raw() {
	nng_socket s;
	int r = nng_pull0_open_raw(&s);
	if( r != 0 ) {
		throw exception(r,"nng_pull0_open_raw");
	}
	return socket(s);
}

}
}}

#endif


<|start_filename|>include/nngpp/http/server.h<|end_filename|>
#ifndef NNGPP_HTTP_SERVER_H
#define NNGPP_HTTP_SERVER_H
#include "server_view.h"
#include <nngpp/url_view.h>

namespace nng { namespace http {

struct server : server_view {

	server() = default;
	
	explicit server( nng_http_server* s ) noexcept : server_view(s) {}
	
	explicit server( url_view u ) {
		int r = nng_http_server_hold( &s, u.get() );
		if( r != 0 ) {
			throw exception(r,"nng_http_server_hold");
		}
	}
	
	server( const server& rhs ) = delete;
	
	server( server&& rhs ) noexcept : server_view(rhs.s) {
		rhs.s = nullptr;
	}
	
	server& operator=( const server& rhs ) = delete;
	
	server& operator=( server&& rhs ) noexcept {
		if( this != &rhs ) {
			if( s != nullptr ) nng_http_server_release(s);
			s = rhs.s;
			rhs.s = nullptr;
		}
		return *this;
	}
	
	~server() {
		if( s != nullptr ) nng_http_server_release(s);
	}
	
	nng_http_server* release() noexcept {
		auto out = s;
		s = nullptr;
		return out;
	}

};

inline server make_server( url_view u ) {
	return server(u);
}

}}

#endif


<|start_filename|>include/nngpp/stream/stream.h<|end_filename|>
#ifndef NNGPP_STREAM_STREAM_H
#define NNGPP_STREAM_STREAM_H
#include "stream_view.h"
#include "dialer.h"
#include "listener.h"

namespace nng { namespace stream {

struct stream : stream_view {

	stream() = default;
	
	explicit stream( nng_stream* s ) noexcept : stream_view(s) {}
	
	stream( const stream& rhs ) = delete;
	
	stream( stream&& rhs ) noexcept : stream_view(rhs.s) {
		rhs.s = nullptr;
	}
	
	stream& operator=( const stream& rhs ) = delete;
	
	stream& operator=( stream&& rhs ) noexcept {
		if( this != &rhs ) {
			if( s != nullptr ) nng_stream_free(s);
			s = rhs.s;
			rhs.s = nullptr;
		}
		return *this;
	}
	
	~stream() {
		if( s != nullptr ) nng_stream_free(s);
	}
	
	nng_stream* release() noexcept {
		auto out = s;
		s = nullptr;
		return out;
	}

};

}}

#endif


<|start_filename|>include/nngpp/nngpp.h<|end_filename|>
#ifndef NNGPP_NNGPP_H
#define NNGPP_NNGPP_H
#include "socket.h"
#include "ctx.h"
#include "dialer.h"
#include "listener.h"
#include "pipe.h"
#include "aio.h"
#include "url.h"
#include "stat.h"
#include "stream/dialer.h"
#include "stream/listener.h"
#include "stream/stream.h"
#endif


<|start_filename|>include/nngpp/option.h<|end_filename|>
#ifndef NNGPP_OPTION_H
#define NNGPP_OPTION_H
#include <nng/nng.h>

namespace nng {

enum option_type {
	bool_,
	int_,
	ms,
	size,
	uint64,
	ptr,
	string,
	addr,
};

enum class option {
	socket_name,        // string
	raw,                // bool
	protocol,           // int
	protocol_name,      // string
	peer,               // int
	peer_name,          // string
	recv_buffer,        // int
	send_buffer,        // int
	recv_fd,            // int
	send_fd,            // int
	recv_timeout,       // ms
	send_timeout,       // ms
	local_address,      // sockaddr
	remote_address,     // sockaddr
	url,                // string
	ttl_max,            // int
	recv_size_max,      // size
	reconnect_time_min, // ms
	reconnect_time_max  // ms
};

namespace detail {

#if __cpp_inline_variables >= 201606
inline
#else
static
#endif
constexpr const char* option_names[] = {
	NNG_OPT_SOCKNAME,
	NNG_OPT_RAW,
	NNG_OPT_PROTO,
	NNG_OPT_PROTONAME,
	NNG_OPT_PEER,
	NNG_OPT_PEERNAME,
	NNG_OPT_RECVBUF,
	NNG_OPT_SENDBUF,
	NNG_OPT_RECVFD,
	NNG_OPT_SENDFD,
	NNG_OPT_RECVTIMEO,
	NNG_OPT_SENDTIMEO,
	NNG_OPT_LOCADDR,
	NNG_OPT_REMADDR,
	NNG_OPT_URL,
	NNG_OPT_MAXTTL,
	NNG_OPT_RECVMAXSZ,
	NNG_OPT_RECONNMINT,
	NNG_OPT_RECONNMAXT
};

#if __cpp_inline_variables >= 201606
inline
#else
static
#endif
constexpr option_type option_types[] = {
	option_type::string,
	option_type::bool_,
	option_type::int_,
	option_type::string,
	option_type::int_,
	option_type::string,
	option_type::int_,
	option_type::int_,
	option_type::int_,
	option_type::int_,
	option_type::ms,
	option_type::ms,
	option_type::addr,
	option_type::addr,
	option_type::string,
	option_type::int_,
	option_type::size,
	option_type::ms,
	option_type::ms
};

}

constexpr const char* to_name( option o ) noexcept {
	return detail::option_names[(int)o];
}

constexpr option_type to_type( option o ) noexcept {
	return detail::option_types[(int)o];
}

}

#endif


<|start_filename|>include/nngpp/msg.h<|end_filename|>
#ifndef NNGPP_MSG_H
#define NNGPP_MSG_H
#include "msg_view.h"

namespace nng {

struct msg : msg_view {

	msg() = default;

	explicit msg( nng_msg* m ) noexcept : msg_view(m) {}
	
	explicit msg( size_t size ) {
		int r = nng_msg_alloc( &m, size );
		if( r != 0 ) {
			throw exception(r,"nng_msg_alloc");
		}
	}
	
	msg( const msg& rhs ) : msg_view() {
		int r = nng_msg_dup(&m,rhs.get());
		if( r != 0 ) {
			throw exception(r,"nng_msg_dup");
		}
	}
	
	msg( msg&& rhs ) noexcept : msg_view(rhs.m) {
		rhs.m = nullptr;
	}
	
	msg& operator=( const msg& rhs ) {
		if( this != &rhs ) {
			if( m != nullptr ) nng_msg_free(m);
			int r = nng_msg_dup(&m,rhs.get());
			if( r != 0 ) {
				throw exception(r,"nng_msg_dup");
			}
		}
		return *this;
	}
	
	msg& operator=( msg&& rhs ) noexcept {
		if( this != &rhs ) {
			if( m != nullptr ) nng_msg_free(m);
			m = rhs.m;
			rhs.m = nullptr;
		}
		return *this;
	}
	
	~msg() {
		if( m != nullptr ) nng_msg_free(m);
	}
	
	nng_msg* release() noexcept {
		auto out = m;
		m = nullptr;
		return out;
	}

};

inline msg make_msg( size_t size ) {
	return msg(size);
}

}

#endif


<|start_filename|>test/tls.cpp<|end_filename|>
#include <catch2/catch.hpp>
#include <nngpp/nngpp.h>
#include <nngpp/protocol/pair1.h>
#include <nngpp/transport/tls.h>
#include <nngpp/transport/tcp.h>
#include <nngpp/platform/platform.h>

namespace {

static const char cert[] =
    "-----BEGIN CERTIFICATE-----\n"
    "MIIDLjCCAhYCCQDtskvsw6K+mDANBgkqhkiG9w0BAQsFADBYMQswCQYDVQQGEwJV\n"
    "UzELMAkGA1UECAwCQ0ExEjAQBgNVBAcMCVNhbiBEaWVnbzEUMBIGA1UECgwLbmFu\n"
    "b21zZy5vcmcxEjAQBgNVBAMMCWxvY2FsaG9zdDAgFw0xODAxMTEyMjM0MzVaGA8y\n"
    "MTE3MTIxODIyMzQzNVowWDELMAkGA1UEBhMCVVMxCzAJBgNVBAgMAkNBMRIwEAYD\n"
    "VQQHDAlTYW4gRGllZ28xFDASBgNVBAoMC25hbm9tc2cub3JnMRIwEAYDVQQDDAls\n"
    "b2NhbGhvc3QwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQDMvoHdEnfO\n"
    "hmG3PTj6YC5qz6N5hgmcwf4EZkor4+R1Q5hDOKqOknWmVuGBD5mA61ObK76vycIT\n"
    "Tp+H+vKvfgunySZrlyYg8IbgoDbvVgj9RF8xFHdN0PVeqnkBCsCzLtSu6TP8PSgI\n"
    "SKiRMH0NUSakWqCPEc2E1r1CKdOpa7av/Na30LPsuKFcAUhu7QiVYfER86ktrO8G\n"
    "F2PeVy44Q8RkiLw8uhU0bpAflqkR1KCjOLajw1eL3C+Io75Io8qUOLxWc3LH0hl3\n"
    "oEI0jWu7JYlRAw/O7xm4pcGTwy5L8Odz4a7ZTAmuapFRarGOIcDg8Yr0tllRd1mH\n"
    "<KEY>tAgMBAAEwDQYJKoZIhvcNAQELBQADggE<KEY>lXUHH\n"
    "<KEY>"
    "<KEY>"
    "<KEY>"
    "<KEY>"
    "dFMXOO1rleU0lWAJcXWOWHH3er0fivu2ISL8fRjjikYvhRGxtkwC0kPDa2Ntzgd3\n"
    "Hsg=\n"
    "-----END CERTIFICATE-----\n";

static const char key[] =
    "-----BEGIN RSA PRIVATE KEY-----\n"
    "<KEY>"
    "jpJ1plbhgQ+ZgOtTmyu+r8nCE06fh/ryr34Lp8kma5cmIPCG4KA271YI/URfMRR3\n"
    "TdD1Xqp5AQrAsy7Urukz/D0oCEiokTB9DVEmpFqgjxHNhNa9QinTqWu2r/zWt9Cz\n"
    "7LihXAFIbu0IlWHxEfOpLazvBhdj3lcuOEPEZIi8PLoVNG6QH5apEdSgozi2o8NX\n"
    "i9wviKO+SKPKlDi8VnNyx9IZd6BCNI1ruyWJUQMPzu8ZuKXBk8MuS/Dnc+Gu2UwJ\n"
    "<KEY>dZh9U+Gdlr+0bNLQIDAQABAoIBAC82HqvjfkzZH98o\n"
    "<KEY>"
    "rDhbUES4VWupxtkBnZheWUyHAjukcG7Y0UnYTTwvAwgCerzWp6RNkfcwAvMmDfis\n"
    "<KEY>cTNv3f7R0cJmaZ/t9FT7SerXf1LW7itvTjRor8/ZK\n"
    "KPwT4oklp1o6RFXSenn/e2e3rAjI+TEwJA3Zp5dqO/M/AhaZKVaxL4voDVdVVkT+\n"
    "LHJWVhjLY5ilPkmPWqmZ2reTaF+gGSSjAQ+t/ahGWFqEdWIz9UoXhBBOd1ibeyvd\n"
    "Kyxp1QECgYEA8KcDkmwPrhqFlQe/U+Md27OhrQ4cecLCa6EVLsCXN1bFyCi3NSo2\n"
    "<KEY>K<KEY>"
    "zbnWH3xgsvvsxdt6hInFhcABLDupW336tJ8EcH7mOKoIP+azwF4kPiUCgYEA2c09\n"
    "zJBUW6SZXhgJ5vgENYc+UwDT7pfhIWZaRL+wXnwSoa7igodTKJtQp/KfFBJK4RA0\n"
    "<KEY>"
    "<KEY>"
    "<KEY>"
    "<KEY>"
    "<KEY>"
    "<KEY>"
    "655UxVIhD3f8Oa/j1lr7acEU5KCiBtkjDU8vOMBsv+FpWOQrlB1JQa/X/+G+bHLF\n"
    "XmUerNkCgYEAv7R8vIKgJ1f69imgHdB31kue3wnOO/6NlfY3GTcaZcTdChY8SZ5B\n"
    "xits8xog0VcaxXhWlfO0hyCnZ9YRQbyDu0qp5eBU2p3qcE01x4ljJBZUOTweG06N\n"
    "cL9dYcwse5FhNMjrQ/OKv6B38SIXpoKQUtjgkaMtmpK8cXX1eqEMNkM=\n"
    "-----END RSA PRIVATE KEY-----\n";


static void init_dialer_tls( nng::dialer_view d ) {
	nng::tls::config cfg( nng::tls::mode::client );
	cfg.config_ca_chain(cert)
		.config_server_name("127.0.0.1")
		.config_auth_mode(nng::tls::auth_mode::none);
	set_opt_config( d, cfg );
}

static void init_listener_tls( nng::listener_view l ) {
	nng::tls::config cfg( nng::tls::mode::server );
	cfg.config_own_cert(cert,key);
	set_opt_config( l, cfg );
}

}

TEST_CASE("We can bind to port zero", "[tls]" ) {
	REQUIRE_NOTHROW(nng::tls::register_transport());

	auto s1 = nng::pair::open();
	REQUIRE(s1);
	auto s2 = nng::pair::open();
	REQUIRE(s2);

	nng::listener l( s1, "tls+tcp://127.0.0.1:0" );
	REQUIRE(l);
	
	REQUIRE_NOTHROW(init_listener_tls(l));
	REQUIRE_NOTHROW(l.start());

	auto addr = get_opt_url(l);
	CAPTURE( addr.data<char>() );

	nng::dialer d( s2, addr.data<char>() );
	REQUIRE_NOTHROW(init_dialer_tls(d));
	REQUIRE_NOTHROW(nng::tls::set_opt_auth_mode(d,nng::tls::auth_mode::none));
	REQUIRE_NOTHROW(d.start());
}

TEST_CASE("Malformed TLS addresses do not panic", "[tls]" ) {
	REQUIRE_NOTHROW(nng::tls::register_transport());

	auto s1 = nng::pair::open();
	REQUIRE(s1);

	auto invalid_address = nng::exception(nng::error::addrinval).what();

	REQUIRE_THROWS_WITH( s1.dial("tls+tcp://127.0.0.1"   ), invalid_address );
	REQUIRE_THROWS_WITH( s1.dial("tls+tcp://127.0.0.1.32"), invalid_address );
	REQUIRE_THROWS_WITH( s1.dial("tls+tcp://127.0.x.1.32"), invalid_address );
	REQUIRE_THROWS_WITH( s1.dial("tls+tcp://127.0.0.1.32"), invalid_address );
	REQUIRE_THROWS_WITH( s1.dial("tls+tcp://127.0.x.1.32"), invalid_address );
}

TEST_CASE("No delay option", "[tls]" ) {
	REQUIRE_NOTHROW(nng::tls::register_transport());

	auto s = nng::pair::open();
	REQUIRE(s);

	REQUIRE( nng::tcp::get_opt_no_delay(s) );
		
	{
		auto d = nng::dialer( s, "tls+tcp://127.0.0.1:4999" );
		REQUIRE( nng::tcp::get_opt_no_delay(d) );

		REQUIRE_NOTHROW( nng::tcp::set_opt_no_delay(d,false) );
		REQUIRE_FALSE( nng::tcp::get_opt_no_delay(d) );

		auto bad_type = nng::exception(nng::error::badtype).what();

		REQUIRE_THROWS_WITH( d.get_opt_int( NNG_OPT_TCP_NODELAY ), bad_type );
		REQUIRE_THROWS_WITH( d.set_opt_int( NNG_OPT_TCP_NODELAY, 0 ), bad_type );
		
		auto inval = nng::exception(nng::error::inval).what();

		int x = 0;
		REQUIRE_THROWS_WITH( d.set_opt( NNG_OPT_TCP_NODELAY, {&x, sizeof(x)} ), inval );

		nng::listener l( s, "tls+tcp://127.0.0.1:4999" );
		REQUIRE( nng::tcp::get_opt_no_delay(l) );
		
		REQUIRE_THROWS_WITH( l.set_opt_int( NNG_OPT_TCP_NODELAY, 0 ), bad_type );

		x = 0;
		REQUIRE_THROWS_WITH( l.set_opt( NNG_OPT_TCP_NODELAY, {&x, sizeof(x)} ), inval );
	}

	// Make sure socket wide defaults apply.
	REQUIRE_NOTHROW( nng::tcp::set_opt_no_delay(s,true) );
	REQUIRE( nng::tcp::get_opt_no_delay(s) );
		
	REQUIRE_NOTHROW( nng::tcp::set_opt_no_delay(s,false) );

	auto d = nng::dialer( s, "tls+tcp://127.0.0.1:4999" );
	REQUIRE_FALSE( nng::tcp::get_opt_no_delay(d) );
}

TEST_CASE("Keepalive option", "[tls]" ) {
	REQUIRE_NOTHROW(nng::tls::register_transport());

	auto s = nng::pair::open();
	REQUIRE(s);

	REQUIRE_FALSE( nng::tcp::get_opt_keep_alive(s) );
	{
		auto d = nng::dialer( s, "tls+tcp://127.0.0.1:4999" );
		REQUIRE_FALSE( nng::tcp::get_opt_keep_alive(d) );

		REQUIRE_NOTHROW( nng::tcp::set_opt_keep_alive(d,true) );
		REQUIRE( nng::tcp::get_opt_keep_alive(d) );

		auto bad_type = nng::exception(nng::error::badtype).what();

		REQUIRE_THROWS_WITH( d.get_opt_int( NNG_OPT_TCP_KEEPALIVE ), bad_type );
		REQUIRE_THROWS_WITH( d.set_opt_int( NNG_OPT_TCP_KEEPALIVE, 1 ), bad_type );

		auto l = nng::listener( s, "tls+tcp://127.0.0.1:4999" );
		REQUIRE_FALSE( nng::tcp::get_opt_keep_alive(l) );
		
		REQUIRE_THROWS_WITH( l.set_opt_int( NNG_OPT_TCP_KEEPALIVE, 1 ), bad_type );
	}

	// Make sure socket wide defaults apply.
	REQUIRE_NOTHROW( nng::tcp::set_opt_keep_alive(s,false) );
	REQUIRE_FALSE( nng::tcp::get_opt_keep_alive(s) );

	REQUIRE_NOTHROW( nng::tcp::set_opt_keep_alive(s,true) );
	auto d = nng::dialer( s, "tls+tcp://127.0.0.1:4999" );
	REQUIRE( nng::tcp::get_opt_keep_alive(d) );
}


<|start_filename|>include/nngpp/protocol/bus0.h<|end_filename|>
#ifndef NNGPP_BUS0_H
#define NNGPP_BUS0_H
#include <nngpp/socket.h>
#include <nng/protocol/bus0/bus.h>

namespace nng { namespace bus {
inline namespace v0 {

inline socket open() {
	nng_socket s;
	int r = nng_bus0_open(&s);
	if( r != 0 ) {
		throw exception(r,"nng_bus0_open");
	}
	return socket(s);
}

inline socket open_raw() {
	nng_socket s;
	int r = nng_bus0_open_raw(&s);
	if( r != 0 ) {
		throw exception(r,"nng_bus0_open_raw");
	}
	return socket(s);
}

}
}}

#endif


<|start_filename|>test/respondpoll.cpp<|end_filename|>
#include <catch2/catch.hpp>
#include <nngpp/nngpp.h>
#include <nngpp/protocol/survey0.h>
#include <nngpp/protocol/respond0.h>
#include <nngpp/platform/platform.h>

TEST_CASE("Respondent pollable") {
	auto addr = "inproc://respondpoll";

	SECTION("Given a connected survey pair") {
		auto surv = nng::survey::open();
		REQUIRE(surv);
		auto resp = nng::respond::open();
		REQUIRE(resp);
		nng::ctx ctx( resp );
		REQUIRE(ctx);

		REQUIRE_NOTHROW( nng::set_opt_send_timeout(surv,2000) );
		REQUIRE_NOTHROW( nng::set_opt_send_timeout(resp,2000) );
		REQUIRE_NOTHROW( nng::set_opt_recv_timeout(surv,2000) );
		REQUIRE_NOTHROW( nng::set_opt_recv_timeout(resp,2000) );

		REQUIRE_NOTHROW( resp.listen(addr) );

		SECTION("Respondent ctx not pollable") {
			REQUIRE_THROWS_WITH(
				ctx.get_opt_int( NNG_OPT_SENDFD ),
				nng::exception(nng::error::notsup).what()
			);
			REQUIRE_THROWS_WITH(
				ctx.get_opt_int( NNG_OPT_RECVFD ),
				nng::exception(nng::error::notsup).what()
			);
		}

		SECTION("Respondent starts not writable") {
			nng::get_opt_send_fd(resp);

			SECTION("And remains unwritable on connect") {
				REQUIRE_NOTHROW( surv.dial(addr) );

				nng::msleep(100);

				SECTION("Becomes writable after recv") {
					surv.send( nng::make_msg(0) );
					resp.recv_msg();
				}
			}
		}

		SECTION("Respondent starts not readable") {
			nng::get_opt_recv_fd(resp);

			SECTION("And doesn't become readable on connect") {
				REQUIRE_NOTHROW( surv.dial(addr) );
				nng::msleep(100);
			}

			SECTION("And becomes readable on data") {
				REQUIRE_NOTHROW( surv.dial(addr) );
				nng::msleep(200);

				auto msg = nng::make_msg(0);
				msg.body().append( {"xyz",3} );
				surv.send( std::move(msg) );

				nng::msleep(300); // give time for msg to arrive
				
				SECTION("Is no longer readable after recv") {
					msg = resp.recv_msg();
				}
			}
		}
	}
}


<|start_filename|>test/tcpsupp.cpp<|end_filename|>
#include <catch2/catch.hpp>
#include <cstring>
#include <nngpp/nngpp.h>
#include <nngpp/transport/tcp.h>

inline uint16_t test_htons( uint16_t in ) noexcept {
	in = ((in >> 8) & 0xff) | ((in & 0xff) << 8);
	return in;
}

TEST_CASE("Supplemental TCP", "[tcpapi]") {
	
	INFO("Listener listens (wildcard)");
	auto l = nng::stream::listener("tcp://127.0.0.1");
	REQUIRE(l);
	REQUIRE_NOTHROW(l.listen());

	uint8_t ip[4];
	ip[0] = 127;
	ip[1] = 0;
	ip[2] = 0;
	ip[3] = 1;

	auto sa = l.get_addr( to_name(nng::option::local_address) );

	REQUIRE(sa.s_in.sa_port != 0);
	REQUIRE(memcmp(&sa.s_in.sa_addr, ip, 4) == 0);

	INFO("We can dial it");
	nng::aio daio(nullptr,nullptr);
	nng::aio laio(nullptr,nullptr);
	nng::aio maio(nullptr,nullptr);

	char uri[64];
	snprintf(uri, sizeof(uri), "tcp://127.0.0.1:%d", test_htons(sa.s_in.sa_port));
	auto d = nng::stream::dialer(uri);
	REQUIRE(d);
	
	d.dial(daio);
	l.accept(laio);

	daio.wait();
	REQUIRE(daio.result() == nng::error::success);
	laio.wait();
	REQUIRE(laio.result() == nng::error::success);

	nng::stream::stream c1( daio.get_output<nng_stream>(0) );
	nng::stream::stream c2( laio.get_output<nng_stream>(0) );
	REQUIRE(c1);
	REQUIRE(c2);
	
	INFO("They exchange messages");
	nng::aio aio1(nullptr,nullptr);
	nng::aio aio2(nullptr,nullptr);

	REQUIRE_NOTHROW(c1.set_bool( to_name(nng::tcp::option::no_delay), true) );
	REQUIRE_NOTHROW(c2.set_bool( to_name(nng::tcp::option::no_delay), true) );
	REQUIRE_NOTHROW(c1.set_bool( to_name(nng::tcp::option::keep_alive), true) );
	
	char buf1[5];
	memcpy(buf1, "TEST", 5);

	char buf2[5];
	memset(buf2, 0, 5);
	
	aio1.set_iov( {buf1, 5} );
	aio2.set_iov( {buf2, 5} );
	c1.send(aio1);
	c2.recv(aio2);
	aio1.wait();
	aio2.wait();

	REQUIRE(aio1.result() == nng::error::success);
	REQUIRE(aio1.count() == 5);

	REQUIRE(aio2.result() == nng::error::success);
	REQUIRE(aio2.count() == 5);

	REQUIRE(memcmp(buf1, buf2, 5) == 0);

	INFO("Socket name matches");
	{
		nng_sockaddr sa2;
		REQUIRE_NOTHROW(sa2 = c2.get_addr( to_name(nng::option::local_address) ));
		REQUIRE(sa2.s_in.sa_family == NNG_AF_INET);
		REQUIRE(sa2.s_in.sa_addr == sa.s_in.sa_addr);
		REQUIRE(sa2.s_in.sa_port == sa.s_in.sa_port);
	}
	
	INFO("Peer name matches")
	{
		nng_sockaddr sa2;
		REQUIRE_NOTHROW(sa2 = c1.get_addr( to_name(nng::option::remote_address) ));
		REQUIRE(sa2.s_in.sa_family == NNG_AF_INET);
		REQUIRE(sa2.s_in.sa_addr == sa.s_in.sa_addr);
		REQUIRE(sa2.s_in.sa_port == sa.s_in.sa_port);
	}

}


<|start_filename|>include/nngpp/protocol/rep0.h<|end_filename|>
#ifndef NNGPP_REP0_H
#define NNGPP_REP0_H
#include <nngpp/socket.h>
#include <nng/protocol/reqrep0/rep.h>

namespace nng { namespace rep {
inline namespace v0 {

inline socket open() {
	nng_socket s;
	int r = nng_rep0_open(&s);
	if( r != 0 ) {
		throw exception(r,"nng_rep0_open");
	}
	return socket(s);
}

inline socket open_raw() {
	nng_socket s;
	int r = nng_rep0_open_raw(&s);
	if( r != 0 ) {
		throw exception(r,"nng_rep0_open_raw");
	}
	return socket(s);
}

}
}}

#endif


<|start_filename|>include/nngpp/pipe.h<|end_filename|>
#ifndef NNGPP_PIPE_H
#define NNGPP_PIPE_H
#include "pipe_view.h"

namespace nng {

struct pipe : pipe_view {

	pipe() = default;

	explicit pipe( nng_pipe p ) noexcept : pipe_view(p) {}
	
	pipe( const pipe& rhs ) = delete;
	
	pipe( pipe&& rhs ) noexcept : pipe_view(rhs.p) {
		rhs.p.id = 0;
	}
	
	pipe& operator=( const pipe& rhs ) = delete;
	
	pipe& operator=( pipe&& rhs ) {
		if( this != &rhs ) {
			if( p.id != 0 ) {
				int r = nng_pipe_close(p);
				if( r != 0 ) {
					throw exception(r,"nng_pipe_close");
				}
			}
			p = rhs.p;
			rhs.p.id = 0;
		}
		return *this;
	}
	
	~pipe() {
		if( p.id != 0 ) nng_pipe_close(p);
	}

	nng_pipe release() noexcept {
		auto out = p;
		p.id = 0;
		return out;
	}

};

}

#endif


<|start_filename|>include/nngpp/platform/mtx.h<|end_filename|>
#ifndef NNGPP_MTX_H
#define NNGPP_MTX_H
#include "mtx_view.h"
#include <nngpp/error.h>

namespace nng {

struct mtx : mtx_view {

	mtx() = default;
	
	explicit mtx( nng_mtx* m ) noexcept : mtx_view(m) {}
	
	mtx( const mtx& rhs ) = delete;
	
	mtx( mtx&& rhs ) noexcept : mtx_view(rhs.m) {
		rhs.m = nullptr;
	}
	
	mtx& operator=( const mtx& rhs ) = delete;
	
	mtx& operator=( mtx&& rhs ) noexcept {
		if( this != &rhs ) {
			if( m != nullptr ) nng_mtx_free(m);
			m = rhs.m;
			rhs.m = nullptr;
		}
		return *this;
	}
	
	~mtx() {
		if( m != nullptr ) nng_mtx_free(m);
	}
	
	nng_mtx* release() noexcept {
		auto out = m;
		m = nullptr;
		return out;
	}

};

inline mtx make_mtx() {
	nng_mtx* m;
	int r = nng_mtx_alloc( &m );
	if( r != 0 ) {
		throw exception(r,"nng_mtx_alloc");
	}
	return mtx(m);
}

}

#endif


<|start_filename|>demo/reqrep/reqrep.cpp<|end_filename|>
// This is a port of the nng demo to nngpp
// See https://github.com/nanomsg/nng/tree/master/demo/reqrep

#include <cstdint>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <ctime>
#include <nngpp/nngpp.h>
#include <nngpp/protocol/req0.h>
#include <nngpp/protocol/rep0.h>

#define CLIENT "client"
#define SERVER "server"
#define DATECMD 1

#define PUT64(ptr, u) \
	do { \
		(ptr)[0] = (uint8_t)(((uint64_t)(u)) >> 56); \
		(ptr)[1] = (uint8_t)(((uint64_t)(u)) >> 48); \
		(ptr)[2] = (uint8_t)(((uint64_t)(u)) >> 40); \
		(ptr)[3] = (uint8_t)(((uint64_t)(u)) >> 32); \
		(ptr)[4] = (uint8_t)(((uint64_t)(u)) >> 24); \
		(ptr)[5] = (uint8_t)(((uint64_t)(u)) >> 16); \
		(ptr)[6] = (uint8_t)(((uint64_t)(u)) >>  8); \
		(ptr)[7] = (uint8_t)((uint64_t)(u)); \
	} while(0)

#define GET64(ptr, v) \
	v = (((uint64_t)((uint8_t)(ptr)[0])) << 56) + \
	    (((uint64_t)((uint8_t)(ptr)[1])) << 48) + \
	    (((uint64_t)((uint8_t)(ptr)[2])) << 40) + \
	    (((uint64_t)((uint8_t)(ptr)[3])) << 32) + \
	    (((uint64_t)((uint8_t)(ptr)[4])) << 24) + \
	    (((uint64_t)((uint8_t)(ptr)[5])) << 16) + \
	    (((uint64_t)((uint8_t)(ptr)[6])) <<  8) + \
	    (((uint64_t)(uint8_t)(ptr)[7]))

void showdate(time_t now) {
	printf("%s", asctime(localtime(&now)));
}

void server(const char* url) {
	auto sock = nng::rep::open();
	sock.listen(url);
	while(true) {
		uint64_t val;
		auto buf = sock.recv();
		if( buf.size() == sizeof(uint64_t) && ((GET64(buf.data<char>(), val)) == DATECMD) ) {
			printf("SERVER: RECEIVED DATE REQUEST\n");
			auto now = time(nullptr);
			printf("SERVER: SENDING DATE: ");
			showdate(now);

			// Reuse the buffer.  We know it is big enough.
			PUT64(buf.data<char>(), (uint64_t)now);
			sock.send( std::move(buf) );
		}
	}
}

void client(const char* url) {
	auto sock = nng::req::open();
	sock.dial(url);

	printf("CLIENT: SENDING DATE REQUEST\n");
	
	uint8_t cmd[sizeof(uint64_t)];
	PUT64(cmd,DATECMD);

	sock.send( cmd );
	auto buf = sock.recv();

	if(buf.size() == sizeof(uint64_t)) {
		uint64_t now;
		GET64(buf.data<char>(), now);
		printf("CLIENT: RECEIVED DATE: ");
		showdate((time_t)now);
	} else {
		printf("CLIENT: GOT WRONG SIZE!\n");
	}
}

int main( int argc, char** argv ) try {
	if(argc > 1 && strcmp(CLIENT, argv[1]) == 0) {
		client(argv[2]);
		return 0;
	}

	if(argc > 1 && strcmp(SERVER, argv[1]) == 0) {
		server(argv[2]);
		return 0;
	}

	fprintf(stderr, "Usage: reqrep %s|%s <URL> ...\n", CLIENT, SERVER);
	return 1;
}
catch( const nng::exception& e ) {
	fprintf(stderr, "%s: %s\n", e.who(), e.what());
	return 1;
}



<|start_filename|>include/nngpp/protocol/survey0.h<|end_filename|>
#ifndef NNGPP_SURVEY0_H
#define NNGPP_SURVEY0_H
#include <nngpp/ctx.h>
#include <nng/protocol/survey0/survey.h>

namespace nng { namespace survey {
inline namespace v0 {

inline socket open() {
	nng_socket s;
	int r = nng_surveyor0_open(&s);
	if( r != 0 ) {
		throw exception(r,"nng_surveyor0_open");
	}
	return socket(s);
}

inline socket open_raw() {
	nng_socket s;
	int r = nng_surveyor0_open_raw(&s);
	if( r != 0 ) {
		throw exception(r,"nng_surveyor0_open_raw");
	}
	return socket(s);
}

enum class option {
	survey_time // ms
};

namespace detail {

#if __cpp_inline_variables >= 201606
inline
#else
static
#endif
const char* const option_names[] = {
	NNG_OPT_SURVEYOR_SURVEYTIME
};

}

inline const char* to_name( option o ) noexcept {
	return detail::option_names[(int)o];
}

inline nng_duration get_opt_survey_time( socket_view s ) {
	return s.get_opt_ms( to_name(option::survey_time) );
}

inline void set_opt_survey_time( socket_view s, nng_duration v ) {
	s.set_opt_ms( to_name(option::survey_time), v );
}

inline nng_duration get_opt_survey_time( ctx_view s ) {
	return s.get_opt_ms( to_name(option::survey_time) );
}

inline void set_opt_survey_time( ctx_view s, nng_duration v ) {
	s.set_opt_ms( to_name(option::survey_time), v );
}

}
}}

#endif


<|start_filename|>include/nngpp/ctx.h<|end_filename|>
#ifndef NNGPP_CTX_H
#define NNGPP_CTX_H
#include "ctx_view.h"
#include "socket_view.h"

namespace nng {

struct ctx : ctx_view {

	ctx() = default;
	
	explicit ctx( nng_ctx c ) noexcept : ctx_view(c) {}
	
	explicit ctx( socket_view s ) {
		int r = nng_ctx_open( &c, s.get() );
		if( r != 0 ) {
			throw exception(r,"nng_ctx_open");
		}
	}
	
	ctx( const ctx& rhs ) = delete;
	
	ctx( ctx&& rhs ) noexcept : ctx_view(rhs.c) {
		rhs.c.id = 0;
	}
	
	ctx& operator=( const ctx& rhs ) = delete;
	
	ctx& operator=( ctx&& rhs ) {
		if( this != &rhs ) {
			if( c.id != 0 ) {
				int r = nng_ctx_close(c);
				if( r != 0 && r != (int)nng::error::closed ) {
					throw exception(r,"nng_ctx_close");
				}
			}
			c = rhs.c;
			rhs.c.id = 0;
		}
		return *this;
	}
	
	~ctx() {
		if( c.id != 0 ) nng_ctx_close(c);
	}
	
	nng_ctx release() noexcept {
		auto out = c;
		c.id = 0;
		return out;
	}
	
};

inline ctx make_ctx( socket_view s ) {
	return ctx(s);
}

}

#endif


<|start_filename|>include/nngpp/socket_view.h<|end_filename|>
#ifndef NNGPP_SOCKET_VIEW_H
#define NNGPP_SOCKET_VIEW_H
#include "core.h"
#include "option.h"
#include "aio_view.h"

namespace nng {

struct socket_view {
protected:
	nng_socket s{0};
	
public:

	socket_view() = default;

	socket_view( nng_socket s ) noexcept : s(s) {}
	
	nng_socket get() const noexcept {
		return s;
	}
	
	explicit operator bool() const noexcept {
		return s.id != 0;
	}
	
	int id() const noexcept {
		return nng_socket_id(s);
	}

	void dial( const char* addr, int flags = 0 ) const {
		int r = nng_dial( s, addr, nullptr, flags );
		if( r != 0 ) {
			throw exception(r,"nng_dial");
		}
	}
	
	void listen( const char* addr, int flags = 0 ) const {
		int r = nng_listen( s, addr, nullptr, flags );
		if( r != 0 ) {
			throw exception(r,"nng_listen");
		}
	}

	void send( view buf, int flags = 0 ) const {
		int r = nng_send(s,buf.data(),buf.size(),flags & ~flag::alloc);
		if( r != 0 ) {
			throw exception(r,"nng_send");
		}
	}

	void send( buffer&& buf, int flags = 0 ) const {
		int r = nng_send(s,buf.data(),buf.size(),flags | flag::alloc);
		if( r != 0 ) {
			throw exception(r,"nng_send");
		}
		// if successful, the buffer has been freed
		buf.release();
	}

	void send( msg&& m, int flags = 0 ) const {
		int r = nng_sendmsg(s,m.get(),flags);
		if( r != 0 ) {
			throw exception(r,"nng_sendmsg");
		}
		// if successful, the msg is owned by the socket
		m.release();
	}

	/**
	 * Send asynchronously
	 */
	void send( aio_view a ) const noexcept {
		nng_send_aio(s,a.get());
	}

	/**
	 * Receives into a pre-allocated buffer
	 * Synchronous
	 * Returns the number of bytes written
	 */
	size_t recv( view out, int flags = 0 ) const {
		size_t sz = out.size();
		int r = nng_recv(s,out.data(),&sz,flags & ~flag::alloc);
		if( r != 0 ) {
			throw exception(r,"nng_recv");
		}
		return sz;
	}

	/**
	 * Receives into a buffer
	 * Synchronous
	 * Allocates and copies
	 */
	buffer recv( int flags = 0 ) const {
		void* data;
		size_t size;
		int r = nng_recv(s,&data,&size,flags | flag::alloc);
		if( r != 0 ) {
			throw exception(r,"nng_recv");
		}
		return buffer(data,size);
	}

	/**
	 * Receive a msg object
	 * Synchronous
	 * No allocation
	 * No copy
	 */
	msg recv_msg( int flags = 0 ) const {
		nng_msg* m;
		int r = nng_recvmsg(s,&m,flags);
		if( r != 0 ) {
			throw exception(r,"nng_recvmsg");
		}
		return msg(m);
	}

	/**
	 * Receive asynchronously
	 */
	void recv( aio_view a ) const noexcept {
		nng_recv_aio(s,a.get());
	}
	
	void pipe_notify( pipe_ev ev, nng_pipe_cb cb, void* arg ) const {
		int r = nng_pipe_notify( s, (nng_pipe_ev)(int)ev, cb, arg );
		if( r != 0 ) {
			throw exception(r,"nng_pipe_notify");
		}
	}
	
	void set_opt( const char* name, view v ) const {
		int r = nng_socket_set(s,name,v.data(),v.size());
		if( r != 0 ) {
			throw exception(r,"nng_socket_set");
		}
	}
	
	void set_opt_bool( const char* name, bool value ) const {
		int r = nng_socket_set_bool(s,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_socket_set_bool");
		}
	}
	
	void set_opt_int( const char* name, int value ) const {
		int r = nng_socket_set_int(s,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_socket_set_int");
		}
	}
	
	void set_opt_size( const char* name, size_t value ) const {
		int r = nng_socket_set_size(s,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_socket_set_size");
		}
	}
	
	void set_opt_uint64( const char* name, uint64_t value ) const {
		int r = nng_socket_set_uint64(s,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_socket_set_uint64");
		}
	}
	
	void set_opt_string( const char* name, const char* value ) const {
		int r = nng_socket_set_string(s,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_socket_set_string");
		}
	}
	
	void set_opt_ptr( const char* name, void* value ) const {
		int r = nng_socket_set_ptr(s,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_socket_set_ptr");
		}
	}
	
	void set_opt_ms( const char* name, nng_duration value ) const {
		int r = nng_socket_set_ms(s,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_socket_set_ms");
		}
	}
	
	void set_opt_addr( const char* name, const nng_sockaddr* value ) const {
		int r = nng_socket_set_addr(s,name,value);
		if( r != 0 ) {
			throw exception(r,"nng_socket_set_addr");
		}
	}
	
	size_t get_opt( const char* name, buffer& out ) const {
		size_t size = out.size();
		int r = nng_socket_get(s,name,out.data(),&size);
		if( r != 0 ) {
			throw exception(r,"nng_socket_get");
		}
		return size;
	}
	
	bool get_opt_bool( const char* name ) const {
		bool out;
		int r = nng_socket_get_bool(s,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_socket_get_bool");
		}
		return out;
	}
	
	int get_opt_int( const char* name ) const {
		int out;
		int r = nng_socket_get_int(s,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_socket_get_int");
		}
		return out;
	}
	
	size_t get_opt_size( const char* name ) const {
		size_t out;
		int r = nng_socket_get_size(s,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_socket_get_size");
		}
		return out;
	}
	
	uint64_t get_opt_uint64( const char* name ) const {
		uint64_t out;
		int r = nng_socket_get_uint64(s,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_socket_get_uint64");
		}
		return out;
	}
	
	buffer get_opt_string( const char* name ) const {
		char* data;
		int r = nng_socket_get_string(s,name,&data);
		if( r != 0 ) {
			throw exception(r,"nng_socket_get_string");
		}
		return buffer(data,strlen(data)+1);
	}

	void* get_opt_ptr( const char* name ) const {
		void* out;
		int r = nng_socket_get_ptr(s,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_socket_get_ptr");
		}
		return out;
	}
	
	nng_duration get_opt_ms( const char* name ) const {
		nng_duration out;
		int r = nng_socket_get_ms(s,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_socket_get_ms");
		}
		return out;
	}

	nng_sockaddr get_opt_addr( const char* name ) const {
		nng_sockaddr out;
		int r = nng_socket_get_addr(s,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_socket_get_addr");
		}
		return out;
	}
	
};

inline socket_view pipe_view::get_socket() const noexcept {
	return nng_pipe_socket(p);
}

inline void device( socket_view lhs, socket_view rhs ) {
	int r = nng_device(lhs.get(),rhs.get());
	if( r != 0 ) {
		throw exception(r,"nng_device");
	}
}

inline buffer get_opt_socket_name( socket_view s ) {
	return s.get_opt_string( to_name(option::socket_name) );
}

inline bool get_opt_raw( socket_view s ) {
	return s.get_opt_bool( to_name(option::raw) );
}

inline int get_opt_protocol( socket_view s ) {
	return s.get_opt_int( to_name(option::protocol) );
}

inline buffer get_opt_protocol_name( socket_view s ) {
	return s.get_opt_string( to_name(option::protocol_name) );
}

inline int get_opt_peer( socket_view s ) {
	return s.get_opt_int( to_name(option::peer) );
}

inline buffer get_opt_peer_name( socket_view s ) {
	return s.get_opt_string( to_name(option::peer_name) );
}

inline int get_opt_recv_buffer( socket_view s ) {
	return s.get_opt_int( to_name(option::recv_buffer) );
}

inline int get_opt_send_buffer( socket_view s ) {
	return s.get_opt_int( to_name(option::send_buffer) );
}

inline int get_opt_recv_fd( socket_view s ) {
	return s.get_opt_int( to_name(option::recv_fd) );
}

inline int get_opt_send_fd( socket_view s ) {
	return s.get_opt_int( to_name(option::send_fd) );
}

inline nng_duration get_opt_recv_timeout( socket_view s ) {
	return s.get_opt_ms( to_name(option::recv_timeout) );
}

inline nng_duration get_opt_send_timeout( socket_view s ) {
	return s.get_opt_ms( to_name(option::send_timeout) );
}

inline int get_opt_ttl_max( socket_view s ) {
	return s.get_opt_int( to_name(option::ttl_max) );
}

inline buffer get_opt_recv_size_max( socket_view s ) {
	return s.get_opt_string( to_name(option::recv_size_max) );
}

inline nng_duration get_opt_reconnect_time_min( socket_view s ) {
	return s.get_opt_ms( to_name(option::reconnect_time_min) );
}

inline nng_duration get_opt_reconnect_time_max( socket_view s ) {
	return s.get_opt_ms( to_name(option::reconnect_time_max) );
}

inline void set_opt_socket_name( socket_view s, const char* v ) {
	s.set_opt_string( to_name(option::socket_name), v );
}

inline void set_opt_recv_buffer( socket_view s, int v ) {
	s.set_opt_int( to_name(option::recv_buffer), v );
}

inline void set_opt_send_buffer( socket_view s, int v ) {
	s.set_opt_int( to_name(option::send_buffer), v );
}

inline void set_opt_recv_timeout( socket_view s, nng_duration v ) {
	s.set_opt_ms( to_name(option::recv_timeout), v );
}

inline void set_opt_send_timeout( socket_view s, nng_duration v ) {
	s.set_opt_ms( to_name(option::send_timeout), v );
}

inline void set_opt_ttl_max( socket_view s, int v ) {
	s.set_opt_int( to_name(option::ttl_max), v );
}
inline void set_opt_reconnect_time_min( socket_view s, nng_duration v ) {
	s.set_opt_ms( to_name(option::reconnect_time_min), v );
}

inline void set_opt_reconnect_time_max( socket_view s, nng_duration v ) {
	s.set_opt_ms( to_name(option::reconnect_time_max), v );
}

}

#endif


<|start_filename|>include/nngpp/stream/listener.h<|end_filename|>
#ifndef NNGPP_STREAM_LISTENER_H
#define NNGPP_STREAM_LISTENER_H
#include "listener_view.h"
#include <nngpp/url_view.h>

namespace nng { namespace stream {

struct listener : listener_view {

	listener() = default;
	
	explicit listener( nng_stream_listener* d ) noexcept : listener_view(d) {}

	explicit listener( const char* url ) {
		int r = nng_stream_listener_alloc( &d, url );
		if( r != 0 ) {
			throw exception(r,"nng_stream_listener_alloc");
		}
	}

	explicit listener( nng::url_view url ) {
		int r = nng_stream_listener_alloc_url( &d, url.get() );
		if( r != 0 ) {
			throw exception(r,"nng_stream_listener_alloc_url");
		}
	}
	
	listener( const listener& rhs ) = delete;
	
	listener( listener&& rhs ) noexcept : listener_view(rhs.d) {
		rhs.d = nullptr;
	}
	
	listener& operator=( const listener& rhs ) = delete;
	
	listener& operator=( listener&& rhs ) noexcept {
		if( this != &rhs ) {
			if( d != nullptr ) nng_stream_listener_free(d);
			d = rhs.d;
			rhs.d = nullptr;
		}
		return *this;
	}
	
	~listener() {
		if( d != nullptr ) nng_stream_listener_free(d);
	}
	
	nng_stream_listener* release() noexcept {
		auto out = d;
		d = nullptr;
		return out;
	}

};

inline listener make_listener( const char* url ) {
	return listener(url);
}

inline listener make_listener( nng::url_view url ) {
	return listener(url);
}

}}

#endif


<|start_filename|>include/nngpp/listener.h<|end_filename|>
#ifndef NNGPP_LISTENER_H
#define NNGPP_LISTENER_H
#include "listener_view.h"
#include "socket_view.h"

namespace nng {

struct listener : listener_view {

	listener() = default;
	
	explicit listener( nng_listener lis ) noexcept : listener_view(lis) {}
	
	explicit listener( socket_view s, const char* addr, int flags ) {
		int r = nng_listen( s.get(), addr, &lis, flags );
		if( r != 0 ) {
			throw exception(r,"nng_listen");
		}
	}
	
	explicit listener( socket_view s, const char* addr ) {
		int r = nng_listener_create( &lis, s.get(), addr );
		if( r != 0 ) {
			throw exception(r,"nng_listener_create");
		}
	}
	
	listener( const listener& rhs ) = delete;
	
	listener( listener&& rhs ) noexcept : listener_view(rhs.lis) {
		rhs.lis.id = 0;
	}
	
	listener& operator=( const listener& rhs ) = delete;
	
	listener& operator=( listener&& rhs ) {
		if( this != &rhs ) {
			if( lis.id != 0 ) {
				int r = nng_listener_close(lis);
				if( r != 0 && r != (int)nng::error::closed ) {
					throw exception(r,"nng_listener_close");
				}
			}
			lis = rhs.lis;
			rhs.lis.id = 0;
		}
		return *this;
	}
	
	~listener() {
		if( lis.id != 0 ) nng_listener_close(lis);
	}

	nng_listener release() noexcept {
		auto out = lis;
		lis.id = 0;
		return out;
	}

};

inline listener make_listener( socket_view s, const char* addr, int flags ) {
	return listener(s,addr,flags);
}

inline listener make_listener( socket_view s, const char* addr ) {
	return listener(s,addr);
}

}

#endif


<|start_filename|>include/nngpp/view.h<|end_filename|>
#ifndef NNGPP_VIEW_H
#define NNGPP_VIEW_H
#include <cstring>
#include "error.h"

namespace nng {

struct view {
protected:
	using pointer = void*;
	pointer d = nullptr;
	size_t s = 0;
	
public:
	view() = default;
	
	template<typename T>
	view( T* d, size_t s ) noexcept : d((pointer)d), s(s) {}

	template<typename T,size_t N>
	view( T(&d)[N] ) noexcept : d((pointer)d), s(N * sizeof(T)) {}

	view( nng_iov x ) noexcept : view(x.iov_buf,x.iov_len) {}
	
	template<typename T = void>
	T* data() const noexcept {
		return (T*)d;
	}
	
	size_t size() const noexcept {
		return s;
	}

	explicit operator bool() const noexcept {
		return d != nullptr;
	}

	bool operator==( const view& rhs ) const noexcept {
		return s == rhs.s && memcmp(d,rhs.d,s) == 0;
	}

	bool operator!=( const view& rhs ) const noexcept {
		return !(*this == rhs);
	}

	operator nng_iov() const noexcept {
		return { d, s };
	}

};

}

#endif


<|start_filename|>perf/local_thr.cpp<|end_filename|>
// This is a port of nng perf to nngpp
// See https://github.com/nanomsg/nng/tree/master/perf

#include <cstdio>
#include "throughput.h"
#include "misc.h"

int main(int argc, char** argv) try {
	if(argc != 4) {
		fprintf(stderr,"Usage: local_thr <listen-addr> <msg-size> <count>");
		return 1;
	}

	auto msgsize = parse_int(argv[2], "Invalid message size");
	auto trips   = parse_int(argv[3], "Invalid count");

	throughput_server(argv[1], msgsize, trips);
}
catch( const nng::exception& e ) {
	fprintf(stderr, "%s: %s\n", e.who(), e.what());
	return 1;
}


<|start_filename|>perf/inproc_thr.cpp<|end_filename|>
// This is a port of nng perf to nngpp
// See https://github.com/nanomsg/nng/tree/master/perf

#include <cstdio>
#include "throughput.h"
#include "inproc.h"
#include "misc.h"

int main(int argc, char** argv) try {
	if(argc != 3) {
		fprintf(stderr,"Usage: inproc_thr <msg-size> <count>");
		return 1;
	}

	inproc_args ia;
	ia.addr    = "inproc://tput_test";
	ia.msgsize = parse_int(argv[1], "Invalid message size");
	ia.count   = parse_int(argv[2], "Invalid count");
	ia.func    = throughput_server;

	nng::thread thr( do_inproc, &ia );

	// Sleep a bit.
	nng::msleep(100);

	throughput_client("inproc://tput_test", ia.msgsize, ia.count);
}
catch( const nng::exception& e ) {
	fprintf(stderr, "%s: %s\n", e.who(), e.what());
	return 1;
}


<|start_filename|>include/nngpp/dialer_view.h<|end_filename|>
#ifndef NNGPP_DIALER_VIEW_H
#define NNGPP_DIALER_VIEW_H
#include "option.h"
#include "buffer.h"

namespace nng {

struct dialer_view {
protected:
	nng_dialer d{0};
	
public:

	dialer_view() = default;
	
	dialer_view( nng_dialer d ) noexcept : d(d) {}
	
	nng_dialer get() const noexcept {
		return d;
	}

	explicit operator bool() const noexcept {
		return d.id != 0;
	}
	
	void start( int flags = 0 ) const {
		int r = nng_dialer_start(d,flags);
		if( r != 0 ) {
			throw exception(r,"nng_dialer_start");
		}
	}
	
	int id() const noexcept {
		return nng_dialer_id(d);
	}
	
	void set_opt( const char* name, view v ) const {
		int r = nng_dialer_set(d,name,v.data(),v.size());
		if( r != 0 ) {
			throw exception(r,"nng_dialer_set");
		}
	}
	
	void set_opt_bool( const char* name, bool val ) const {
		int r = nng_dialer_set_bool(d,name,val);
		if( r != 0 ) {
			throw exception(r,"nng_dialer_set_bool");
		}
	}
	
	void set_opt_int( const char* name, int val ) const {
		int r = nng_dialer_set_int(d,name,val);
		if( r != 0 ) {
			throw exception(r,"nng_dialer_set_int");
		}
	}
	
	void set_opt_size( const char* name, size_t val ) const {
		int r = nng_dialer_set_size(d,name,val);
		if( r != 0 ) {
			throw exception(r,"nng_dialer_set_size");
		}
	}
	
	void set_opt_uint64( const char* name, uint64_t val ) const {
		int r = nng_dialer_set_uint64(d,name,val);
		if( r != 0 ) {
			throw exception(r,"nng_dialer_set_uint64");
		}
	}
	
	void set_opt_string( const char* name, const char* val ) const {
		int r = nng_dialer_set_string(d,name,val);
		if( r != 0 ) {
			throw exception(r,"nng_dialer_set_string");
		}
	}
	
	void set_opt_ptr( const char* name, void* val ) const {
		int r = nng_dialer_set_ptr(d,name,val);
		if( r != 0 ) {
			throw exception(r,"nng_dialer_set_ptr");
		}
	}
	
	void set_opt_ms( const char* name, nng_duration val ) const {
		int r = nng_dialer_set_ms(d,name,val);
		if( r != 0 ) {
			throw exception(r,"nng_dialer_set_ms");
		}
	}
	
	void set_opt_addr( const char* name, const nng_sockaddr* val ) const {
		int r = nng_dialer_set_addr(d,name,val);
		if( r != 0 ) {
			throw exception(r,"nng_dialer_set_addr");
		}
	}

	size_t get_opt( const char* name, buffer& out ) const {
		size_t size = out.size();
		int r = nng_dialer_get(d,name,out.data(),&size);
		if( r != 0 ) {
			throw exception(r,"nng_dialer_get");
		}
		return size;
	}
	
	bool get_opt_bool( const char* name ) const {
		bool out;
		int r = nng_dialer_get_bool(d,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_dialer_get_bool");
		}
		return out;
	}
	
	int get_opt_int( const char* name ) const {
		int out;
		int r = nng_dialer_get_int(d,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_dialer_get_int");
		}
		return out;
	}
	
	size_t get_opt_size( const char* name ) const {
		size_t out;
		int r = nng_dialer_get_size(d,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_dialer_get_size");
		}
		return out;
	}
	
	uint64_t get_opt_uint64( const char* name ) const {
		uint64_t out;
		int r = nng_dialer_get_uint64(d,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_dialer_get_uint64");
		}
		return out;
	}
	
	buffer get_opt_string( const char* name ) const {
		char* data;
		int r = nng_dialer_get_string(d,name,&data);
		if( r != 0 ) {
			throw exception(r,"nng_dialer_get_string");
		}
		return buffer(data,strlen(data)+1);
	}
	
	void* get_opt_ptr( const char* name ) const {
		void* out;
		int r = nng_dialer_get_ptr(d,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_dialer_get_ptr");
		}
		return out;
	}
	
	nng_duration get_opt_ms( const char* name ) const {
		nng_duration out;
		int r = nng_dialer_get_ms(d,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_dialer_get_ms");
		}
		return out;
	}
	
	nng_sockaddr get_opt_addr( const char* name ) const {
		nng_sockaddr out;
		int r = nng_dialer_get_addr(d,name,&out);
		if( r != 0 ) {
			throw exception(r,"nng_dialer_get_addr");
		}
		return out;
	}
	
};

inline buffer get_opt_url( dialer_view s ) {
	return s.get_opt_string( to_name(option::url) );
}

inline size_t get_opt_recv_size_max( dialer_view s ) {
	return s.get_opt_size( to_name(option::recv_size_max) );
}

inline nng_duration get_opt_reconnect_time_min( dialer_view s ) {
	return s.get_opt_ms( to_name(option::reconnect_time_min) );
}

inline nng_duration get_opt_reconnect_time_max( dialer_view s ) {
	return s.get_opt_ms( to_name(option::reconnect_time_max) );
}

inline void set_opt_recv_size_max( dialer_view s, size_t v ) {
	s.set_opt_size( to_name(option::recv_size_max), v );
}

inline void set_opt_reconnect_time_min( dialer_view s, nng_duration v ) {
	s.set_opt_ms( to_name(option::reconnect_time_min), v );
}

inline void set_opt_reconnect_time_max( dialer_view s, nng_duration v ) {
	s.set_opt_ms( to_name(option::reconnect_time_max), v );
}

}

#endif


<|start_filename|>test/device.cpp<|end_filename|>
#include <catch2/catch.hpp>
#include <nngpp/nngpp.h>
#include <nngpp/protocol/pair1.h>
#include <nngpp/platform/platform.h>
#include <thread>

TEST_CASE("We cannot create cooked mode device","[device]") {
	auto s1 = nng::pair::v1::open();
	REQUIRE(s1);
	REQUIRE_THROWS_WITH(
		nng::device(s1,s1),
		nng::exception(nng::error::inval).what()
	);
}

TEST_CASE("PAIRv1 device","[device]") {
	auto addr1 = "inproc://dev1";
	auto addr2 = "inproc://dev2";

	SECTION("We can create a PAIRv1 device") {
		std::thread thr;
		{
			auto dev1 = nng::pair::v1::open_raw();
			REQUIRE(dev1);
			auto dev2 = nng::pair::v1::open_raw();
			REQUIRE(dev2);

			thr = std::thread([&] {
				try { nng::device(dev1,dev2); }
				catch(...) {}
			});

			REQUIRE_NOTHROW( dev1.listen(addr1) );
			REQUIRE_NOTHROW( dev2.listen(addr2) );

			auto end1 = nng::pair::v1::open();
			REQUIRE(end1);
			auto end2 = nng::pair::v1::open();
			REQUIRE(end2);

			REQUIRE_NOTHROW( end1.dial(addr1) );
			REQUIRE_NOTHROW( end2.dial(addr2) );

			nng_duration tmo = 1000;
			REQUIRE_NOTHROW( nng::set_opt_recv_timeout(end1,tmo) );
			REQUIRE_NOTHROW( nng::set_opt_recv_timeout(end2,tmo) );

			nng::msleep(100);
			SECTION("Device can send and receive") {
				{
					auto msg = nng::make_msg(0);
					REQUIRE(msg);

					msg.body().append("ALPHA");
					REQUIRE_NOTHROW( end1.send(std::move(msg)) );
					REQUIRE_NOTHROW( msg = end2.recv_msg() );
					REQUIRE( msg.body().get() == "ALPHA" );
				}
				{
					auto msg = nng::make_msg(0);
					REQUIRE(msg);

					msg.body().append("OMEGA");
					REQUIRE_NOTHROW( end2.send(std::move(msg)) );
					REQUIRE_NOTHROW( msg = end1.recv_msg() );
					REQUIRE( msg.body().get() == "OMEGA" );
				}
			}
		}
		thr.join();
	}
}


<|start_filename|>include/nngpp/platform/thread.h<|end_filename|>
#ifndef NNGPP_THREAD_H
#define NNGPP_THREAD_H
#include "thread_view.h"
#include <nngpp/error.h>

namespace nng {

struct thread : thread_view {

	thread() = default;
	
	explicit thread( nng_thread* t ) noexcept : thread_view(t) {}

	explicit thread( void(*f)(void*), void* arg ) {
		int r = nng_thread_create( &t, f, arg );
		if( r != 0 ) {
			throw exception(r,"nng_thread_create");
		}
	}

	thread( const thread& rhs ) = delete;
	
	thread( thread&& rhs ) noexcept : thread_view(rhs.t) {
		rhs.t = nullptr;
	}
	
	thread& operator=( const thread& rhs ) = delete;
	
	thread& operator=( thread&& rhs ) noexcept {
		if( this != &rhs ) {
			if( t != nullptr ) nng_thread_destroy(t);
			t = rhs.t;
			rhs.t = nullptr;
		}
		return *this;
	}
	
	~thread() {
		if( t != nullptr ) nng_thread_destroy(t);
	}
	
	nng_thread* release() noexcept {
		auto out = t;
		t = nullptr;
		return out;
	}

};

inline thread make_thread( void(*f)(void*), void* arg ) {
	return thread(f,arg);
}

}

#endif


<|start_filename|>include/nngpp/http/client.h<|end_filename|>
#ifndef NNGPP_HTTP_CLIENT_H
#define NNGPP_HTTP_CLIENT_H
#include "client_view.h"
#include <nngpp/url_view.h>

namespace nng { namespace http {

struct client : client_view {

	client() = default;
	
	explicit client( nng_http_client* c ) noexcept : client_view(c) {}
	
	explicit client( url_view u ) {
		int r = nng_http_client_alloc( &c, u.get() );
		if( r != 0 ) {
			throw exception(r,"nng_http_client_alloc");
		}
	}
	
	client( const client& rhs ) = delete;
	
	client( client&& rhs ) noexcept : client_view(rhs.c) {
		rhs.c = nullptr;
	}
	
	client& operator=( const client& rhs ) = delete;
	
	client& operator=( client&& rhs ) noexcept {
		if( this != &rhs ) {
			if( c != nullptr ) nng_http_client_free(c);
			c = rhs.c;
			rhs.c = nullptr;
		}
		return *this;
	}
	
	~client() {
		if( c != nullptr ) nng_http_client_free(c);
	}
	
	nng_http_client* release() noexcept {
		auto out = c;
		c = nullptr;
		return out;
	}

};

inline client make_client( url_view u ) {
	return client(u);
}

}}

#endif


<|start_filename|>include/nngpp/platform/mtx_view.h<|end_filename|>
#ifndef NNGPP_MTX_VIEW_H
#define NNGPP_MTX_VIEW_H
#include <nng/nng.h>
#include <nng/supplemental/util/platform.h>

namespace nng {

struct mtx_view {
protected:
	nng_mtx* m = nullptr;
	
public:
	mtx_view() = default;
	
	mtx_view( nng_mtx* m ) noexcept : m(m) {}

	nng_mtx* get() const noexcept {
		return m;
	}

	nng_mtx* operator->() const noexcept {
		return m;
	}

	explicit operator bool() const noexcept {
		return m != nullptr;
	}

	void lock() const noexcept {
		nng_mtx_lock(m);
	}

	void unlock() const noexcept {
		nng_mtx_unlock(m);
	}

};

}

#endif


<|start_filename|>include/nngpp/http/http.h<|end_filename|>
#ifndef NNGPP_HTTP_H
#define NNGPP_HTTP_H
#include "client.h"
#include "conn.h"
#include "handler.h"
#include "req.h"
#include "res.h"
#include "server.h"
#endif


<|start_filename|>perf/misc.h<|end_filename|>
#ifndef NNGPP_PERF_MISC_H
#define NNGPP_PERF_MISC_H
#include <cstdlib>
#include <cstdarg>
#include <cstdio>
#include <stdexcept>

static int parse_int(const char* arg, const char* what) {
	char* eptr;
	auto val = strtol(arg, &eptr, 10);
	// Must be a postive number less than around a billion.
	if((val < 0) || (val > (1 << 30)) || (*eptr != 0) || (eptr == arg)) {
		throw std::runtime_error(what);
	}
	return (int)val;
}

static void die(const char* fmt, ...) {
	va_list ap;
	va_start(ap, fmt);
	vfprintf(stderr, fmt, ap);
	va_end(ap);
	fprintf(stderr, "\n");
	exit(2);
}

#endif


<|start_filename|>test/main.cpp<|end_filename|>
#define CATCH_CONFIG_RUNNER
#define CATCH_CONFIG_COLOUR_NONE
#include <catch2/catch.hpp>
#include <nngpp/nngpp.h>

int main( int argc, char* argv[] ) {
	int result = Catch::Session().run( argc, argv );
	nng::fini();
	return result;
}


<|start_filename|>include/nngpp/http/res.h<|end_filename|>
#ifndef NNGPP_HTTP_RES_H
#define NNGPP_HTTP_RES_H
#include "res_view.h"

namespace nng { namespace http {

struct res : res_view {

	res() = default;
	
	explicit res( nng_http_res* s ) noexcept : res_view(s) {}
	
	explicit res( status code ) {
		int r = nng_http_res_alloc_error( &s, (uint16_t)code );
		if( r != 0 ) {
			throw exception(r,"nng_http_res_alloc_error");
		}
	}
	
	res( const res& rhs ) = delete;
	
	res( res&& rhs ) noexcept : res_view(rhs.s) {
		rhs.s = nullptr;
	}
	
	res& operator=( const res& rhs ) = delete;
	
	res& operator=( res&& rhs ) noexcept {
		if( this != &rhs ) {
			if( s != nullptr ) nng_http_res_free(s);
			s = rhs.s;
			rhs.s = nullptr;
		}
		return *this;
	}
	
	~res() {
		if( s != nullptr ) nng_http_res_free(s);
	}
	
	nng_http_res* release() noexcept {
		auto out = s;
		s = nullptr;
		return out;
	}

};

inline res make_res() {
	nng_http_res* s;
	int r = nng_http_res_alloc( &s );
	if( r != 0 ) {
		throw exception(r,"nng_http_res_alloc");
	}
	return res(s);
}

inline res make_res( status code ) {
	return res(code);
}

}}

#endif


<|start_filename|>test/reconnect.cpp<|end_filename|>
#include <catch2/catch.hpp>
#include <nngpp/nngpp.h>
#include <nngpp/protocol/push0.h>
#include <nngpp/protocol/pull0.h>
#include <nngpp/platform/platform.h>
#include <cstring>

TEST_CASE("Reconnect works","[reconnect]") {
	auto addr = "inproc://reconnect";

	SECTION("Dialing before listening works") {
		auto push = nng::push::open();
		REQUIRE(push);
		auto pull = nng::pull::open();
		REQUIRE(pull);

		REQUIRE_NOTHROW( nng::set_opt_reconnect_time_min( pull, 10 ) );
		REQUIRE_NOTHROW( nng::set_opt_reconnect_time_max( pull, 10 ) );

		REQUIRE_NOTHROW( push.dial(addr,nng::flag::nonblock) );
		nng::msleep(100);
		REQUIRE_NOTHROW( pull.listen(addr) );

		SECTION("We can send a frame") {
			nng::msleep(100);

			auto msg = nng::make_msg(0);
			REQUIRE(msg);
			msg.body().append("hello");
			REQUIRE_NOTHROW( push.send( std::move(msg) ) );
			REQUIRE_NOTHROW( msg = pull.recv_msg() );

			REQUIRE(msg);
			REQUIRE( msg.body().get() == "hello" );
		}
	}

	SECTION("Reconnection works") {
		auto push = nng::push::open();
		REQUIRE(push);
		auto pull = nng::pull::open();
		REQUIRE(pull);

		REQUIRE_NOTHROW( nng::set_opt_reconnect_time_min( pull, 10 ) );
		REQUIRE_NOTHROW( nng::set_opt_reconnect_time_max( pull, 10 ) );

		REQUIRE_NOTHROW( push.dial(addr,nng::flag::nonblock) );
		{
			nng::listener l( pull, addr, 0 );
			REQUIRE(l);
			nng::msleep(100);
		}
		pull.listen(addr);

		SECTION("They still exchange frames") {
			nng::pipe_view p1;
			nng::msleep(100);
			{
				auto msg = nng::make_msg(0);
				REQUIRE(msg);
				msg.body().append("hello");
				REQUIRE_NOTHROW( push.send( std::move(msg) ) );
				REQUIRE_NOTHROW( msg = pull.recv_msg() );
				REQUIRE(msg);
				REQUIRE( msg.body().get() == "hello" );
				
				p1 = msg.get_pipe();
			}
			SECTION("Even after pipe close") {
				nng::pipe_view p2;
				nng_pipe_close(p1.get());
				nng::msleep(100);
				{
					auto msg = nng::make_msg(0);
					REQUIRE(msg);
					msg.body().append("again");
					REQUIRE_NOTHROW( push.send( std::move(msg) ) );
					REQUIRE_NOTHROW( msg = pull.recv_msg() );
					REQUIRE(msg);
					REQUIRE( msg.body().get() == "again" );

					p2 = msg.get_pipe();
				}
				REQUIRE(p2.id() != p1.id());
			}
		}
	}
}


<|start_filename|>demo/async/client.cpp<|end_filename|>
// This is a port of the nng demo to nngpp
// See https://github.com/nanomsg/nng/tree/master/demo/async

#include <cstdio>
#include <cstdlib>
#include <nngpp/nngpp.h>
#include <nngpp/protocol/req0.h>
#include <nngpp/platform/platform.h>

//  The client runs just once, and then returns.
void client(const char* url, const char* msecstr) {
	auto msec = atoi(msecstr);

	auto sock = nng::req::open();
	sock.dial(url);

	auto start = nng::clock();
	auto msg = nng::make_msg(0);
	msg.body().append_u32(msec);
	sock.send(std::move(msg));
	msg = sock.recv_msg();
	auto end = nng::clock();

	printf("Request took %u milliseconds.\n", (uint32_t)(end - start));
}

int main(int argc, char** argv) try {
	if (argc != 3) {
		fprintf(stderr, "Usage: %s <url> <secs>\n", argv[0]);
		return 1;
	}
	client(argv[1], argv[2]);
}
catch( const nng::exception& e ) {
	fprintf(stderr, "%s: %s\n", e.who(), e.what());
	return 1;
}


<|start_filename|>include/nngpp/protocol/respond0.h<|end_filename|>
#ifndef NNGPP_RESPOND0_H
#define NNGPP_RESPOND0_H
#include <nngpp/socket.h>
#include <nng/protocol/survey0/respond.h>

namespace nng { namespace respond {
inline namespace v0 {

inline socket open() {
	nng_socket s;
	int r = nng_respondent0_open(&s);
	if( r != 0 ) {
		throw exception(r,"nng_respondent0_open");
	}
	return socket(s);
}

inline socket open_raw() {
	nng_socket s;
	int r = nng_respondent0_open_raw(&s);
	if( r != 0 ) {
		throw exception(r,"nng_respondent0_open_raw");
	}
	return socket(s);
}

}
}}

#endif
