summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/socket_base.cpp69
-rw-r--r--src/socket_base.hpp10
2 files changed, 74 insertions, 5 deletions
diff --git a/src/socket_base.cpp b/src/socket_base.cpp
index 3141517..ac3b4b9 100644
--- a/src/socket_base.cpp
+++ b/src/socket_base.cpp
@@ -26,12 +26,18 @@
#include "err.hpp"
#include "zmq_listener.hpp"
#include "io_thread.hpp"
+#include "config.hpp"
zmq::socket_base_t::socket_base_t (app_thread_t *parent_) :
object_t (parent_),
pending_term_acks (0),
- app_thread (parent_)
-{
+ app_thread (parent_),
+ hwm (0),
+ lwm (0),
+ swap (0),
+ mask (0),
+ affinity (0)
+{
}
zmq::socket_base_t::~socket_base_t ()
@@ -63,14 +69,67 @@ zmq::socket_base_t::~socket_base_t ()
int zmq::socket_base_t::setsockopt (int option_, void *optval_,
size_t optvallen_)
{
- zmq_assert (false);
+ switch (option_) {
+
+ case ZMQ_HWM:
+ if (optvallen_ != sizeof (int64_t)) {
+ errno = EINVAL;
+ return -1;
+ }
+ hwm = *((int64_t*) optval_);
+ return 0;
+
+ case ZMQ_LWM:
+ if (optvallen_ != sizeof (int64_t)) {
+ errno = EINVAL;
+ return -1;
+ }
+ lwm = *((int64_t*) optval_);
+ return 0;
+
+ case ZMQ_SWAP:
+ if (optvallen_ != sizeof (int64_t)) {
+ errno = EINVAL;
+ return -1;
+ }
+ swap = *((int64_t*) optval_);
+ return 0;
+
+ case ZMQ_MASK:
+ if (optvallen_ != sizeof (int64_t)) {
+ errno = EINVAL;
+ return -1;
+ }
+ mask = (uint64_t) *((int64_t*) optval_);
+ return 0;
+
+ case ZMQ_AFFINITY:
+ if (optvallen_ != sizeof (int64_t)) {
+ errno = EINVAL;
+ return -1;
+ }
+ affinity = (uint64_t) *((int64_t*) optval_);
+ return 0;
+
+ case ZMQ_SESSIONID:
+ if (optvallen_ != sizeof (const char*)) {
+ errno = EINVAL;
+ return -1;
+ }
+ session_id = (const char*) optval_;
+ return 0;
+
+ default:
+ errno = EINVAL;
+ return -1;
+ }
}
int zmq::socket_base_t::bind (const char *addr_)
{
// TODO: The taskset should be taken from socket options.
- uint64_t taskset = 0;
- zmq_listener_t *listener = new zmq_listener_t (choose_io_thread (taskset), this);
+ zmq_listener_t *listener =
+ new zmq_listener_t (choose_io_thread (affinity), this);
int rc = listener->set_address (addr_);
if (rc != 0)
return -1;
diff --git a/src/socket_base.hpp b/src/socket_base.hpp
index f028c18..7f1c803 100644
--- a/src/socket_base.hpp
+++ b/src/socket_base.hpp
@@ -21,9 +21,11 @@
#define __ZMQ_SOCKET_BASE_HPP_INCLUDED__
#include <set>
+#include <string>
#include "i_api.hpp"
#include "object.hpp"
+#include "stdint.hpp"
namespace zmq
{
@@ -64,6 +66,14 @@ namespace zmq
// Application thread the socket lives in.
class app_thread_t *app_thread;
+ // Socket options.
+ int64_t hwm;
+ int64_t lwm;
+ int64_t swap;
+ uint64_t mask;
+ uint64_t affinity;
+ std::string session_id;
+
socket_base_t (const socket_base_t&);
void operator = (const socket_base_t&);
};