LCOV - code coverage report
Current view: top level - lib/cc - session.cc (source / functions) Hit Total Coverage
Test: report.info Lines: 129 203 63.5 %
Date: 2010-12-08 Functions: 23 42 54.8 %
Legend: Lines: hit not hit | Branches: + taken - not taken # not executed Branches: 55 168 32.7 %

           Branch data     Line data    Source code
       1                 :            : // Copyright (C) 2009  Internet Systems Consortium, Inc. ("ISC")
       2                 :            : //
       3                 :            : // Permission to use, copy, modify, and/or distribute this software for any
       4                 :            : // purpose with or without fee is hereby granted, provided that the above
       5                 :            : // copyright notice and this permission notice appear in all copies.
       6                 :            : //
       7                 :            : // THE SOFTWARE IS PROVIDED "AS IS" AND ISC DISCLAIMS ALL WARRANTIES WITH
       8                 :            : // REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
       9                 :            : // AND FITNESS.  IN NO EVENT SHALL ISC BE LIABLE FOR ANY SPECIAL, DIRECT,
      10                 :            : // INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
      11                 :            : // LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE
      12                 :            : // OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
      13                 :            : // PERFORMANCE OF THIS SOFTWARE.
      14                 :            : 
      15                 :            : // $Id: session.cc 2803 2010-08-25 22:22:16Z jinmei $
      16                 :            : 
      17                 :            : #include <config.h>
      18                 :            : #include <cc/session_config.h>
      19                 :            : 
      20                 :            : #include <stdint.h>
      21                 :            : 
      22                 :            : // XXX: there seems to be a strange dependency between ASIO and std library
      23                 :            : // definitions.  On some platforms if we include std headers before ASIO
      24                 :            : // headers unexpected behaviors will happen.
      25                 :            : // A middle term solution is to generalize our local wrapper interface
      26                 :            : // (currently only available for the auth server), where all such portability
      27                 :            : // issues are hidden, and to have other modules use the wrapper.
      28                 :            : #include <unistd.h>             // for some IPC/network system calls
      29                 :            : #include <asio.hpp>
      30                 :            : #include <asio/error_code.hpp>
      31                 :            : #include <asio/deadline_timer.hpp>
      32                 :            : #include <asio/system_error.hpp>
      33                 :            : 
      34                 :            : #include <cstdio>
      35                 :            : #include <vector>
      36                 :            : #include <iostream>
      37                 :            : #include <sstream>
      38                 :            : 
      39                 :            : #include <sys/un.h>
      40                 :            : 
      41                 :            : #include <boost/bind.hpp>
      42                 :            : #include <boost/optional.hpp>
      43                 :            : #include <boost/function.hpp>
      44                 :            : #include <boost/date_time/posix_time/posix_time_types.hpp>
      45                 :            : 
      46                 :            : #include <exceptions/exceptions.h>
      47                 :            : 
      48                 :            : #include <cc/data.h>
      49                 :            : #include <cc/session.h>
      50                 :            : 
      51                 :            : using namespace std;
      52                 :            : using namespace isc::cc;
      53                 :            : using namespace isc::data;
      54                 :            : 
      55                 :            : // some of the asio names conflict with socket API system calls
      56                 :            : // (e.g. write(2)) so we don't import the entire asio namespace.
      57                 :            : using asio::io_service;
      58                 :            : 
      59                 :            : namespace {
      60                 :            : /// \brief Sets the given Optional 'result' to the given error code
      61                 :            : /// Used as a callback for emulating sync reads with async calls
      62                 :            : /// \param result Pointer to the optional to set
      63                 :            : /// \param err The error code to set it to
      64                 :            : void
      65                 :            : setResult(boost::optional<asio::error_code>* result,
      66                 :         67 :           const asio::error_code& err)
      67                 :            : {
      68                 :         67 :     result->reset(err);
      69                 :         67 : }
      70                 :            : }
      71                 :            : 
      72                 :            : namespace isc {
      73                 :            : namespace cc {
      74                 :            : 
      75                 :          7 : class SessionImpl {
      76                 :            : public:
      77                 :          7 :     SessionImpl(io_service& io_service) :
      78                 :            :         sequence_(-1), queue_(Element::createList()),
      79                 :            :         io_service_(io_service), socket_(io_service_), data_length_(0),
      80                 :          7 :         timeout_(MSGQ_DEFAULT_TIMEOUT)
      81                 :          7 :     {}
      82                 :            :     void establish(const char& socket_file);
      83                 :            :     void disconnect();
      84                 :            :     void writeData(const void* data, size_t datalen);
      85                 :            :     size_t readDataLength();
      86                 :            :     // Blocking read. Will throw a SessionTimeout if the timeout value
      87                 :            :     // (in seconds) is thrown. If timeout is 0 it will block forever
      88                 :            :     void readData(void* data, size_t datalen);
      89                 :            :     void startRead(boost::function<void()> user_handler);
      90                 :          3 :     void setTimeout(size_t seconds) { timeout_ = seconds; };
      91                 :        102 :     size_t getTimeout() const { return timeout_; };
      92                 :            : 
      93                 :            :     long int sequence_; // the next sequence number to use
      94                 :            :     std::string lname_;
      95                 :            :     ElementPtr queue_;
      96                 :            : 
      97                 :            : private:
      98                 :            :     void internalRead(const asio::error_code& error,
      99                 :            :                       size_t bytes_transferred);
     100                 :            : 
     101                 :            : private:
     102                 :            :     io_service& io_service_;
     103                 :            :     asio::local::stream_protocol::socket socket_;
     104                 :            :     uint32_t data_length_;
     105                 :            :     boost::function<void()> user_handler_;
     106                 :            :     asio::error_code error_;
     107                 :            :     size_t timeout_;
     108                 :            : 
     109                 :            :     // By default, unless changed or disabled, blocking reads on
     110                 :            :     // the msgq channel will time out after 4 seconds in this
     111                 :            :     // implementation.
     112                 :            :     // This number is chosen to be low enough so that whatever
     113                 :            :     // component is blocking does not seem to be hanging, but
     114                 :            :     // still gives enough time for other modules to respond if they
     115                 :            :     // are busy. If this choice turns out to be a bad one, we can
     116                 :            :     // change it later.
     117                 :            :     static const size_t MSGQ_DEFAULT_TIMEOUT = 4000;
     118                 :            : };
     119                 :            : 
     120                 :            : void
     121                 :          7 : SessionImpl::establish(const char& socket_file) {
     122                 :            :     try {
     123                 :            :         socket_.connect(asio::local::stream_protocol::endpoint(&socket_file),
     124                 :         13 :                         error_);
     125                 :          2 :     } catch(const asio::system_error& se) {
     126                 :          3 :         isc_throw(SessionError, se.what());
     127                 :            :     }
     128         [ -  + ]:          6 :     if (error_) {
     129                 :          0 :         isc_throw(SessionError, "Unable to connect to message queue: " <<
     130                 :            :                   error_.message());
     131                 :            :     }
     132                 :          6 : }
     133                 :            : 
     134                 :            : void
     135                 :          2 : SessionImpl::disconnect() {
     136                 :          2 :     socket_.close();
     137                 :          2 :     data_length_ = 0;
     138                 :          2 : }
     139                 :            : 
     140                 :            : void
     141                 :         18 : SessionImpl::writeData(const void* data, size_t datalen) {
     142                 :            :     try {
     143                 :         18 :         asio::write(socket_, asio::buffer(data, datalen));
     144                 :          0 :     } catch (const asio::system_error& asio_ex) {
     145                 :          0 :         isc_throw(SessionError, "ASIO write failed: " << asio_ex.what());
     146                 :            :     }
     147                 :         18 : }
     148                 :            : 
     149                 :            : size_t
     150                 :         15 : SessionImpl::readDataLength() {
     151                 :         15 :     size_t ret_len = data_length_;
     152                 :            :     
     153         [ +  + ]:         15 :     if (ret_len == 0) {
     154                 :         11 :         readData(&data_length_, sizeof(data_length_));
     155         [ -  + ]:          8 :         if (data_length_ == 0) {
     156                 :          0 :             isc_throw(SessionError, "ASIO read: data length is not ready");
     157                 :            :         }
     158         [ -  + ]:          8 :         ret_len = ntohl(data_length_);
     159                 :            :     }
     160                 :            : 
     161                 :         12 :     data_length_ = 0;
     162                 :         12 :     return (ret_len);
     163                 :            : }
     164                 :            : 
     165                 :            : void
     166                 :         35 : SessionImpl::readData(void* data, size_t datalen) {
     167                 :            :     boost::optional<asio::error_code> read_result;
     168                 :            :     boost::optional<asio::error_code> timer_result;
     169                 :            : 
     170                 :            :     try {
     171                 :            :         asio::async_read(socket_, asio::buffer(data, datalen),
     172                 :         70 :                          boost::bind(&setResult, &read_result, _1));
     173                 :         35 :         asio::deadline_timer timer(socket_.io_service());
     174                 :            :     
     175         [ +  + ]:         35 :         if (getTimeout() != 0) {
     176                 :         32 :             timer.expires_from_now(boost::posix_time::milliseconds(getTimeout()));
     177                 :         64 :             timer.async_wait(boost::bind(&setResult, &timer_result, _1));
     178                 :            :         }
     179                 :            : 
     180                 :            :         // wait until either we have read the data we want, the
     181                 :            :         // timer expires, or one of the two is triggered with an error.
     182                 :            :         // When one of them has a result, cancel the other, and wait
     183                 :            :         // until the cancel is processed before we continue
     184 [ +  + ][ +  - ]:         81 :         while (!read_result && !timer_result) {
                 [ +  + ]
     185                 :         46 :             socket_.io_service().run_one();
     186                 :            : 
     187                 :            :             // Don't cancel the timer if we haven't set it
     188 [ +  + ][ +  + ]:         46 :             if (read_result && getTimeout() != 0) {
                 [ +  + ]
     189                 :            :                 timer.cancel();
     190         [ +  + ]:         60 :                 while (!timer_result) {
     191                 :         30 :                     socket_.io_service().run_one();
     192                 :            :                 }
     193         [ +  + ]:         16 :             } else if (timer_result) {
     194                 :          2 :                 socket_.cancel();
     195         [ +  + ]:          4 :                 while (!read_result) {
     196                 :          2 :                     socket_.io_service().run_one();
     197                 :            :                 }
     198                 :            :             }
     199                 :            :         }
     200                 :            : 
     201                 :            :         // asio::error_code evaluates to false if there was no error
     202         [ +  + ]:         35 :         if (*read_result) {
     203         [ +  + ]:          3 :             if (*read_result == asio::error::operation_aborted) {
     204                 :          6 :                 isc_throw(SessionTimeout,
     205                 :            :                           "Timeout while reading data from cc session");
     206                 :            :             } else {
     207                 :          3 :                 isc_throw(SessionError,
     208                 :            :                           "Error while reading data from cc session: " <<
     209                 :            :                           read_result->message());
     210                 :            :             }
     211                 :            :         }
     212                 :          0 :     } catch (const asio::system_error& asio_ex) {
     213                 :            :         // to hide ASIO specific exceptions, we catch them explicitly
     214                 :            :         // and convert it to SessionError.
     215                 :          0 :         isc_throw(SessionError, "ASIO read failed: " << asio_ex.what());
     216                 :            :     }
     217                 :         32 : }
     218                 :            : 
     219                 :            : void
     220                 :          4 : SessionImpl::startRead(boost::function<void()> user_handler) {
     221                 :          4 :     data_length_ = 0;
     222                 :          4 :     user_handler_ = user_handler;
     223                 :            :     asio::async_read(socket_, asio::buffer(&data_length_,
     224                 :            :                                            sizeof(data_length_)),
     225                 :            :                      boost::bind(&SessionImpl::internalRead, this,
     226                 :            :                                  asio::placeholders::error,
     227                 :         12 :                                  asio::placeholders::bytes_transferred));
     228                 :          4 : }
     229                 :            : 
     230                 :            : void
     231                 :            : SessionImpl::internalRead(const asio::error_code& error,
     232                 :          4 :                           size_t bytes_transferred)
     233                 :            : {
     234         [ +  - ]:          4 :     if (!error) {
     235         [ -  + ]:          4 :         assert(bytes_transferred == sizeof(data_length_));
     236         [ -  + ]:          4 :         data_length_ = ntohl(data_length_);
     237         [ -  + ]:          4 :         if (data_length_ == 0) {
     238                 :          0 :             isc_throw(SessionError, "Invalid message length (0)");
     239                 :            :         }
     240                 :          4 :         user_handler_();
     241                 :            :     } else {
     242                 :          0 :         isc_throw(SessionError, "asynchronous read failed");
     243                 :            :     }
     244                 :          3 : }
     245                 :            : 
     246                 :          7 : Session::Session(io_service& io_service) : impl_(new SessionImpl(io_service))
     247                 :          7 : {}
     248                 :            : 
     249                 :          7 : Session::~Session() {
     250 [ #  # ][ +  - ]:          7 :     delete impl_;
                 [ #  # ]
     251 [ #  # ][ -  + ]:          7 : }
                 [ #  # ]
     252                 :            : 
     253                 :            : void
     254                 :          1 : Session::disconnect() {
     255                 :          1 :     impl_->disconnect();
     256                 :          1 : }
     257                 :            : 
     258                 :            : void
     259                 :          4 : Session::startRead(boost::function<void()> read_callback) {
     260                 :          4 :     impl_->startRead(read_callback);
     261                 :          4 : }
     262                 :            : 
     263                 :            : namespace {                     // maybe unnecessary.
     264                 :            : // This is a helper class to make the establish() method (below) exception-safe
     265                 :            : // with the RAII approach.
     266                 :            : class SessionHolder {
     267                 :            : public:
     268                 :          6 :     SessionHolder(SessionImpl* obj) : impl_obj_(obj) {}
     269                 :            :     ~SessionHolder()
     270                 :            :     {
     271   [ +  -  -  + ]:          6 :         if (impl_obj_ != NULL) {
     272                 :          1 :             impl_obj_->disconnect();
     273                 :            :         }
     274                 :            :     }
     275                 :          5 :     void clear() { impl_obj_ = NULL; }
     276                 :            :     SessionImpl* impl_obj_;
     277                 :            : };
     278                 :            : }
     279                 :            : 
     280                 :            : void
     281                 :          7 : Session::establish(const char* socket_file) {
     282         [ -  + ]:          7 :     if (socket_file == NULL) {
     283                 :          0 :         socket_file = getenv("BIND10_MSGQ_SOCKET_FILE");
     284                 :            :     }
     285         [ -  + ]:          7 :     if (socket_file == NULL) {
     286                 :          0 :         socket_file = BIND10_MSGQ_SOCKET_FILE;
     287                 :            :     }
     288                 :            : 
     289                 :          7 :     impl_->establish(*socket_file);
     290                 :            : 
     291                 :            :     // once established, encapsulate the implementation object so that we
     292                 :            :     // can safely release the internal resource when exception happens
     293                 :            :     // below.
     294                 :          6 :     SessionHolder session_holder(impl_);
     295                 :            : 
     296                 :            :     //
     297                 :            :     // send a request for our local name, and wait for a response
     298                 :            :     //
     299                 :            :     ConstElementPtr get_lname_msg =
     300                 :         12 :         Element::fromJSON("{ \"type\": \"getlname\" }");
     301                 :          6 :     sendmsg(get_lname_msg);
     302                 :            : 
     303                 :            :     ConstElementPtr routing, msg;
     304                 :          6 :     recvmsg(routing, msg, false);
     305                 :            : 
     306                 :         15 :     impl_->lname_ = msg->get("lname")->stringValue();
     307                 :            : 
     308                 :            :     // At this point there's no risk of resource leak.
     309                 :            :     session_holder.clear();
     310                 :          5 : }
     311                 :            : 
     312                 :            : //
     313                 :            : // Convert to wire format and send this via the stream socket with its length
     314                 :            : // prefix.
     315                 :            : //
     316                 :            : void
     317                 :          6 : Session::sendmsg(ConstElementPtr msg) {
     318                 :          6 :     std::string header_wire = msg->toWire();
     319                 :          6 :     unsigned int length = 2 + header_wire.length();
     320         [ -  + ]:          6 :     unsigned int length_net = htonl(length);
     321                 :          6 :     unsigned short header_length = header_wire.length();
     322         [ -  + ]:          6 :     unsigned short header_length_net = htons(header_length);
     323                 :            : 
     324                 :          6 :     impl_->writeData(&length_net, sizeof(length_net));
     325                 :          6 :     impl_->writeData(&header_length_net, sizeof(header_length_net));
     326                 :          6 :     impl_->writeData(header_wire.data(), header_length);
     327                 :          6 : }
     328                 :            : 
     329                 :            : void
     330                 :          0 : Session::sendmsg(ConstElementPtr env, ConstElementPtr msg) {
     331                 :          0 :     std::string header_wire = env->toWire();
     332                 :          0 :     std::string body_wire = msg->toWire();
     333                 :          0 :     unsigned int length = 2 + header_wire.length() + body_wire.length();
     334         [ #  # ]:          0 :     unsigned int length_net = htonl(length);
     335                 :          0 :     unsigned short header_length = header_wire.length();
     336         [ #  # ]:          0 :     unsigned short header_length_net = htons(header_length);
     337                 :            : 
     338                 :          0 :     impl_->writeData(&length_net, sizeof(length_net));
     339                 :          0 :     impl_->writeData(&header_length_net, sizeof(header_length_net));
     340                 :          0 :     impl_->writeData(header_wire.data(), header_length);
     341                 :          0 :     impl_->writeData(body_wire.data(), body_wire.length());
     342                 :          0 : }
     343                 :            : 
     344                 :            : bool
     345                 :          0 : Session::recvmsg(ConstElementPtr& msg, bool nonblock, int seq) {
     346                 :            :     ConstElementPtr l_env;
     347                 :          0 :     return (recvmsg(l_env, msg, nonblock, seq));
     348                 :            : }
     349                 :            : 
     350                 :            : bool
     351                 :            : Session::recvmsg(ConstElementPtr& env, ConstElementPtr& msg,
     352                 :         15 :                  bool nonblock, int seq)
     353                 :            : {
     354                 :         15 :     size_t length = impl_->readDataLength();
     355         [ -  + ]:         12 :     if (hasQueuedMsgs()) {
     356                 :            :         ConstElementPtr q_el;
     357         [ #  # ]:          0 :         for (int i = 0; i < impl_->queue_->size(); i++) {
     358                 :          0 :             q_el = impl_->queue_->get(i);
     359 [ #  # ][ #  # ]:          0 :             if (( seq == -1 &&
         [ #  # ][ #  # ]
         [ #  # ][ #  # ]
         [ #  # ][ #  # ]
         [ #  # ][ #  # ]
         [ #  # ][ #  # ]
         [ #  # ][ #  # ]
         [ #  # ][ #  # ]
         [ #  # ][ #  # ]
         [ #  # ][ #  # ]
         [ #  # ][ #  # ]
         [ #  # ][ #  # ]
                 [ #  # ]
     360                 :            :                   !q_el->get(0)->contains("reply")
     361                 :            :                 ) || (
     362                 :            :                   q_el->get(0)->contains("reply") &&
     363                 :            :                   q_el->get(0)->get("reply")->intValue() == seq
     364                 :            :                 )
     365                 :            :                ) {
     366                 :          0 :                    env = q_el->get(0);
     367                 :          0 :                    msg = q_el->get(1);
     368                 :          0 :                    impl_->queue_->remove(i);
     369                 :          0 :                    return (true);
     370                 :            :             }
     371                 :            :         }
     372                 :            :     }
     373                 :            :     
     374                 :            :     unsigned short header_length_net;
     375                 :         12 :     impl_->readData(&header_length_net, sizeof(header_length_net));
     376                 :            : 
     377         [ -  + ]:         12 :     unsigned short header_length = ntohs(header_length_net);
     378         [ -  + ]:         12 :     if (header_length > length || length < 2) {
     379                 :          0 :         isc_throw(SessionError, "Length parameters invalid: total=" << length
     380                 :            :                   << ", header=" << header_length);
     381                 :            :     }
     382                 :            : 
     383                 :            :     // remove the header-length bytes from the total length
     384                 :         12 :     length -= 2;
     385                 :         12 :     std::vector<char> buffer(length);
     386                 :         12 :     impl_->readData(&buffer[0], length);
     387                 :            : 
     388                 :         12 :     std::string header_wire = std::string(&buffer[0], header_length);
     389                 :            :     std::string body_wire = std::string(&buffer[0] + header_length,
     390                 :         12 :                                         length - header_length);
     391                 :         12 :     std::stringstream header_wire_stream;
     392                 :         12 :     header_wire_stream << header_wire;
     393                 :            :     ConstElementPtr l_env =
     394                 :         12 :         Element::fromWire(header_wire_stream, header_length);
     395                 :            :     
     396                 :         12 :     std::stringstream body_wire_stream;
     397                 :         12 :     body_wire_stream << body_wire;
     398                 :            :     ConstElementPtr l_msg =
     399                 :         12 :         Element::fromWire(body_wire_stream, length - header_length);
     400 [ +  - ][ -  + ]:         24 :     if ((seq == -1 &&
         [ #  # ][ #  # ]
         [ -  + ][ #  # ]
         [ #  # ][ -  + ]
         [ #  # ][ -  + ]
         [ #  # ][ -  + ]
         [ #  # ][ -  + ]
         [ #  # ][ +  - ]
         [ #  # ][ +  - ]
                 [ +  - ]
     401                 :            :          !l_env->contains("reply")
     402                 :            :         ) || (
     403                 :            :          l_env->contains("reply") &&
     404                 :            :          l_env->get("reply")->intValue() == seq
     405                 :            :         )
     406                 :            :        ) {
     407                 :         12 :         env = l_env;
     408                 :         12 :         msg = l_msg;
     409                 :         12 :         return (true);
     410                 :            :     } else {
     411                 :          0 :         ElementPtr q_el = Element::createList();
     412                 :          0 :         q_el->add(l_env);
     413                 :          0 :         q_el->add(l_msg);
     414                 :          0 :         impl_->queue_->add(q_el);
     415                 :          0 :         return (recvmsg(env, msg, nonblock, seq));
     416                 :         24 :     }
     417                 :            :     // XXXMLG handle non-block here, and return false for short reads
     418                 :            : }
     419                 :            : 
     420                 :            : void
     421                 :          0 : Session::subscribe(std::string group, std::string instance) {
     422                 :          0 :     ElementPtr env = Element::createMap();
     423                 :            : 
     424                 :          0 :     env->set("type", Element::create("subscribe"));
     425                 :          0 :     env->set("group", Element::create(group));
     426                 :          0 :     env->set("instance", Element::create(instance));
     427                 :            : 
     428                 :          0 :     sendmsg(env);
     429                 :          0 : }
     430                 :            : 
     431                 :            : void
     432                 :          0 : Session::unsubscribe(std::string group, std::string instance) {
     433                 :          0 :     ElementPtr env = Element::createMap();
     434                 :            : 
     435                 :          0 :     env->set("type", Element::create("unsubscribe"));
     436                 :          0 :     env->set("group", Element::create(group));
     437                 :          0 :     env->set("instance", Element::create(instance));
     438                 :            : 
     439                 :          0 :     sendmsg(env);
     440                 :          0 : }
     441                 :            : 
     442                 :            : int
     443                 :            : Session::group_sendmsg(ConstElementPtr msg, std::string group,
     444                 :          0 :                        std::string instance, std::string to)
     445                 :            : {
     446                 :          0 :     ElementPtr env = Element::createMap();
     447                 :          0 :     long int nseq = ++impl_->sequence_;
     448                 :            :     
     449                 :          0 :     env->set("type", Element::create("send"));
     450                 :          0 :     env->set("from", Element::create(impl_->lname_));
     451                 :          0 :     env->set("to", Element::create(to));
     452                 :          0 :     env->set("group", Element::create(group));
     453                 :          0 :     env->set("instance", Element::create(instance));
     454                 :          0 :     env->set("seq", Element::create(nseq));
     455                 :            :     //env->set("msg", Element::create(msg->toWire()));
     456                 :            : 
     457                 :          0 :     sendmsg(env, msg);
     458                 :          0 :     return (nseq);
     459                 :            : }
     460                 :            : 
     461                 :            : bool
     462                 :            : Session::group_recvmsg(ConstElementPtr& envelope, ConstElementPtr& msg,
     463                 :          9 :                        bool nonblock, int seq)
     464                 :            : {
     465                 :          9 :     return (recvmsg(envelope, msg, nonblock, seq));
     466                 :            : }
     467                 :            : 
     468                 :            : int
     469                 :          0 : Session::reply(ConstElementPtr envelope, ConstElementPtr newmsg) {
     470                 :          0 :     ElementPtr env = Element::createMap();
     471                 :          0 :     long int nseq = ++impl_->sequence_;
     472                 :            :     
     473                 :          0 :     env->set("type", Element::create("send"));
     474                 :          0 :     env->set("from", Element::create(impl_->lname_));
     475                 :          0 :     env->set("to", Element::create(envelope->get("from")->stringValue()));
     476                 :          0 :     env->set("group", Element::create(envelope->get("group")->stringValue()));
     477                 :          0 :     env->set("instance", Element::create(envelope->get("instance")->stringValue()));
     478                 :          0 :     env->set("seq", Element::create(nseq));
     479                 :          0 :     env->set("reply", Element::create(envelope->get("seq")->intValue()));
     480                 :            : 
     481                 :          0 :     sendmsg(env, newmsg);
     482                 :            : 
     483                 :          0 :     return (nseq);
     484                 :            : }
     485                 :            : 
     486                 :            : bool
     487                 :         12 : Session::hasQueuedMsgs() const {
     488                 :         12 :     return (impl_->queue_->size() > 0);
     489                 :            : }
     490                 :            : 
     491                 :            : void
     492                 :          3 : Session::setTimeout(size_t milliseconds) {
     493                 :          3 :     impl_->setTimeout(milliseconds);
     494                 :          3 : }
     495                 :            : 
     496                 :            : size_t
     497                 :          2 : Session::getTimeout() const {
     498                 :          2 :     return (impl_->getTimeout());
     499                 :            : }
     500                 :            : }
     501 [ +  - ][ +  - ]:         12 : }

Generated by: LCOV version 1.9