3 files changed, 75 insertions, 5 deletions

diff --git a/include/zmq.h b/include/zmq.h
index 63c8434..0bada1e 100644
--- a/include/zmq.h
+++ b/include/zmq.h
@@ -162,6 +162,7 @@ ZMQ_EXPORT void *zmq_socket (void *context, int type);
 ZMQ_EXPORT int zmq_close (void *s);
 
 //  Sets an option on the socket.
+//  EINVAL - unknown option, a value with incorrect length or an invalid value.
 ZMQ_EXPORT int zmq_setsockopt (void *s, int option_, void *optval_,
     size_t optvallen_); 
 
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&);
     };