3party/libzmq
0
0
Fork 0
mirror of https://github.com/zeromq/libzmq.git synced 2025-03-09 15:26:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1297 from rodgert/master

Browse Source 
Remove ZMQ_IDENTITY_FD socket option
This commit is contained in:
Constantin Rack 2015-01-10 00:44:45 +01:00
commit 95e7611471
13 changed files with 13 additions and 187 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
Makefile.am
View File

@ -334,7 +334,6 @@ test_apps = \
tests/test_connect_rid \
tests/test_bind_src_address \
tests/test_metadata \
tests/test_id2fd \
tests/test_capabilities \
tests/test_xpub_nodrop \
tests/test_xpub_manual \

17
doc/zmq_getsockopt.txt
View File

@ -676,23 +676,6 @@ Option value unit:: N/A
Default value:: not set
Applicable socket types:: all, when using TCP transport
ZMQ_IDENTITY_FD: Retrieve FD associated with given identity
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
The 'ZMQ_IDENTITY_FD' option shall retrieve the FD associated with given identity.
call _zmq_getsockopt()_ with _option_value_ / _option_len_ pointing to memory
holding the identity string. On return the start of _option_value_ buffer will be
filled with file descriptor of the pipe with given identity if found. If the identity
is not found ENOTSOCK is returned as _zmq_getsockopt()_ result. When the pipe is not
using FD as lower transport you might get -1 as FD. NB: _option_value_ must be always
big enough to hold sizeof(fd_t) bytes no matter how small the identity length is.
[horizontal]
Option value type:: character string/fd_t
Option value unit:: N/A
Default value:: not set
Applicable socket types:: ROUTER
RETURN VALUE
------------

1
include/zmq.h
View File

@ -292,7 +292,6 @@ ZMQ_EXPORT const char *zmq_msg_gets (zmq_msg_t *msg, const char *property);
#define ZMQ_GSSAPI_SERVICE_PRINCIPAL 64
#define ZMQ_GSSAPI_PLAINTEXT 65
#define ZMQ_HANDSHAKE_IVL 66
#define ZMQ_IDENTITY_FD 67
#define ZMQ_SOCKS_PROXY 68
#define ZMQ_XPUB_NODROP 69
#define ZMQ_BLOCKY 70

7
src/i_engine.hpp
View File

@ -20,8 +20,6 @@
#ifndef __ZMQ_I_ENGINE_HPP_INCLUDED__
#define __ZMQ_I_ENGINE_HPP_INCLUDED__
#include "fd.hpp"
namespace zmq
{
@ -49,10 +47,7 @@ namespace zmq
// are messages to send available.
virtual void restart_output () = 0;
virtual void zap_msg_available () = 0;
// provide a way to link from engine to file descriptor
virtual fd_t get_assoc_fd () { return retired_fd;};
virtual void zap_msg_available () = 0;
};
}

1
src/pipe.cpp
View File

@ -65,7 +65,6 @@ int zmq::pipepair (class object_t *parents_ [2], class pipe_t* pipes_ [2],
zmq::pipe_t::pipe_t (object_t *parent_, upipe_t *inpipe_, upipe_t *outpipe_,
int inhwm_, int outhwm_, bool conflate_) :
object_t (parent_),
assoc_fd (retired_fd),
inpipe (inpipe_),
outpipe (outpipe_),
in_active (true),

3
src/pipe.hpp
View File

@ -27,7 +27,6 @@
#include "stdint.hpp"
#include "array.hpp"
#include "blob.hpp"
#include "fd.hpp"
namespace zmq
{
@ -120,8 +119,6 @@ namespace zmq
// check HWM
bool check_hwm () const;
// provide a way to link pipe to engine fd. Set on session initialization
fd_t assoc_fd; //=retired_fd
private:
// Type of the underlying lock-free pipe.

53
src/router.cpp
View File

@ -34,7 +34,7 @@ zmq::router_t::router_t (class ctx_t *parent_, uint32_t tid_, int sid_) :
next_rid (generate_random ()),
mandatory (false),
// raw_sock functionality in ROUTER is deprecated
raw_sock (false),
raw_sock (false),
probe_router (false),
handover (false)
{
@ -118,8 +118,8 @@ int zmq::router_t::xsetsockopt (int option_, const void *optval_,
return 0;
}
break;
case ZMQ_ROUTER_HANDOVER:
case ZMQ_ROUTER_HANDOVER:
if (is_int && value >= 0) {
handover = (value != 0);
return 0;
@ -133,39 +133,6 @@ int zmq::router_t::xsetsockopt (int option_, const void *optval_,
return -1;
}
int zmq::router_t::xgetsockopt (int option_, const void *optval_,
size_t *optvallen_)
{
switch (option_) {
case ZMQ_IDENTITY_FD:
if (optval_==NULL && optvallen_) {
*optvallen_=sizeof(fd_t);
return 0;
}
if (optval_ && optvallen_ && *optvallen_) {
if (*optvallen_ < sizeof(fd_t)) {
*optvallen_=sizeof(fd_t);
return EINVAL;
}
blob_t identity= blob_t((unsigned char*)optval_,*optvallen_);
outpipes_t::iterator it = outpipes.find (identity);
if (it == outpipes.end() ){
return ENOTSOCK;
}
*((fd_t*)optval_)=it->second.pipe->assoc_fd;
*optvallen_=sizeof(fd_t);
return 0;
}
break;
default:
break;
}
errno = EINVAL;
return -1;
}
void zmq::router_t::xpipe_terminated (pipe_t *pipe_)
{
@ -426,10 +393,10 @@ bool zmq::router_t::identify_peer (pipe_t *pipe_)
connect_rid.length());
connect_rid.clear ();
outpipes_t::iterator it = outpipes.find (identity);
if (it != outpipes.end ())
if (it != outpipes.end ())
zmq_assert(false); // Not allowed to duplicate an existing rid
}
else
else
if (options.raw_sock) { // Always assign identity for raw-socket
unsigned char buf [5];
buf [0] = 0;
@ -437,7 +404,7 @@ bool zmq::router_t::identify_peer (pipe_t *pipe_)
identity = blob_t (buf, sizeof buf);
}
else
if (!options.raw_sock) {
if (!options.raw_sock) {
// Pick up handshake cases and also case where next identity is set
msg.init ();
ok = pipe_->read (&msg);
@ -463,7 +430,7 @@ bool zmq::router_t::identify_peer (pipe_t *pipe_)
return false;
else {
// We will allow the new connection to take over this
// identity. Temporarily assign a new identity to the
// identity. Temporarily assign a new identity to the
// existing pipe so we can terminate it asynchronously.
unsigned char buf [5];
buf [0] = 0;
@ -471,13 +438,13 @@ bool zmq::router_t::identify_peer (pipe_t *pipe_)
blob_t new_identity = blob_t (buf, sizeof buf);
it->second.pipe->set_identity (new_identity);
outpipe_t existing_outpipe =
outpipe_t existing_outpipe =
{it->second.pipe, it->second.active};
ok = outpipes.insert (outpipes_t::value_type (
new_identity, existing_outpipe)).second;
zmq_assert (ok);
// Remove the existing identity entry to allow the new
// connection to take the identity.
outpipes.erase (it);

1
src/router.hpp
View File

@ -47,7 +47,6 @@ namespace zmq
// Overrides of functions from socket_base_t.
void xattach_pipe (zmq::pipe_t *pipe_, bool subscribe_to_all_);
int xsetsockopt (int option_, const void *optval_, size_t optvallen_);
int xgetsockopt (int option_, const void *optval_, size_t *optvallen_);
int xsend (zmq::msg_t *msg_);
int xrecv (zmq::msg_t *msg_);
bool xhas_in ();

4
src/session_base.cpp
View File

@ -367,9 +367,7 @@ void zmq::session_base_t::process_attach (i_engine *engine_)
// Remember the local end of the pipe.
zmq_assert (!pipe);
pipe = pipes [0];
// Store engine assoc_fd for linking pipe to fd
pipe->assoc_fd = engine_->get_assoc_fd ();
pipes [1]->assoc_fd = pipe->assoc_fd;
// Ask socket to plug into the remote end of the pipe.
send_bind (socket, pipes [1]);
}

10
src/socket_base.cpp
View File

@ -289,11 +289,6 @@ int zmq::socket_base_t::getsockopt (int option_, void *optval_,
errno = ETERM;
return -1;
}
// First, check whether specific socket type overloads the option.
int rc = xgetsockopt (option_, optval_, optvallen_);
if (rc == 0 || errno != EINVAL)
return rc;
if (option_ == ZMQ_RCVMORE) {
if (*optvallen_ < sizeof (int)) {
@ -1067,11 +1062,6 @@ int zmq::socket_base_t::xsetsockopt (int, const void *, size_t)
errno = EINVAL;
return -1;
}
int zmq::socket_base_t::xgetsockopt (int, const void *, size_t*)
{
errno = EINVAL;
return -1;
}
bool zmq::socket_base_t::xhas_out ()
{

5
src/socket_base.hpp
View File

@ -133,13 +133,10 @@ namespace zmq
// The default implementation assumes there are no specific socket
// options for the particular socket type. If not so, override this
// methods.
// method.
virtual int xsetsockopt (int option_, const void *optval_,
size_t optvallen_);
virtual int xgetsockopt (int option_, const void *optval_,
size_t *optvallen_);
// The default implementation assumes that send is not supported.
virtual bool xhas_out ();
virtual int xsend (zmq::msg_t *msg_);

2
src/stream_engine.hpp
View File

@ -76,8 +76,6 @@ namespace zmq
void out_event ();
void timer_event (int id_);
// export s via i_engine so it is possible to link a pipe to fd
fd_t get_assoc_fd (){ return s; };
private:
// Unplug the engine from the session.

95
tests/test_id2fd.cpp
View File

@ -1,95 +0,0 @@
/*
Copyright (c) 2007-2014 Contributors as noted in the AUTHORS file
This file is part of 0MQ.
0MQ is free software; you can redistribute it and/or modify it under
the terms of the GNU Lesser General Public License as published by
the Free Software Foundation; either version 3 of the License, or
(at your option) any later version.
0MQ is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "testutil.hpp"
int main (void)
{
setup_test_environment();
void *ctx = zmq_ctx_new ();
assert (ctx);
void *client = zmq_socket (ctx, ZMQ_REQ);
assert (client);
void *server = zmq_socket (ctx, ZMQ_ROUTER);
assert (server);
// Now do a basic ping test
int rc = zmq_bind (server, "tcp://127.0.0.1:9998");
assert (rc == 0);
rc = zmq_connect (client, "tcp://127.0.0.1:9998");
assert (rc == 0);
rc=zmq_send(client,"1234567890",10,0);
assert (rc != -1);
int partnumber=1;
int recvfd=-1;
zmq_msg_t part;
do {
/* if not first free prev message part */
if (partnumber!=1) zmq_msg_close (&part);
/* Create an empty ØMQ message to hold the message part */
int rc = zmq_msg_init (&part);
assert (rc == 0);
/* Block until a message is available to be received from socket */
rc = zmq_msg_recv (&part,server, 0);
assert (rc != -1);
if (partnumber==1) {// this is the identity of the receiving pipe
//buffer for zmq_getsockopt / ZMQ_IDENTITY_FD
char idbuf[255];
char failbuf[2];
size_t idbufsz=zmq_msg_size (&part);
size_t failsz=2;
assert (idbufsz<=255);
memcpy(idbuf,zmq_msg_data(&part),idbufsz);
failbuf[0] = idbuf[0];
failbuf[1] = 0;
// ensure that we validate buffer is sufficient to hold result
rc = zmq_getsockopt (server, ZMQ_IDENTITY_FD, failbuf, &failsz);
assert (rc == EINVAL);
rc = zmq_getsockopt (server, ZMQ_IDENTITY_FD, idbuf, &idbufsz);
assert (rc == 0);
memcpy(&recvfd,idbuf,sizeof(recvfd));
//depending on your system this should be around 14
assert (recvfd > 0);
}
partnumber++;
} while (zmq_msg_more(&part));
zmq_msg_close (&part);
close_zero_linger (client);
close_zero_linger (server);
zmq_ctx_term (ctx);
return 0 ;
}