ceph/src/jaegertracing/thrift/contrib/zeromq/TZmqServer.h

   1 /*
   2  * Licensed to the Apache Software Foundation (ASF) under one
   3  * or more contributor license agreements. See the NOTICE file
   4  * distributed with this work for additional information
   5  * regarding copyright ownership. The ASF licenses this file
   6  * to you under the Apache License, Version 2.0 (the
   7  * "License"); you may not use this file except in compliance
   8  * with the License. You may obtain a copy of the License at
   9  *
  10  *   http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing,
  13  * software distributed under the License is distributed on an
  14  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  15  * KIND, either express or implied. See the License for the
  16  * specific language governing permissions and limitations
  17  * under the License.
  18  */
  19
  20 #ifndef _THRIFT_SERVER_TZMQSERVER_H_
  21 #define _THRIFT_SERVER_TZMQSERVER_H_ 1
  22
  23 #include <memory>
  24 #include <zmq.hpp>
  25 #include <thrift/server/TServer.h>
  26
  27 namespace apache { namespace thrift { namespace server {
  28
  29 class TZmqServer : public TServer {
  30  public:
  31   TZmqServer(
  32       std::shared_ptr<TProcessor> processor,
  33       zmq::context_t& ctx, const std::string& endpoint, int type)
  34     : TServer(processor)
  35     , processor_(processor)
  36     , zmq_type_(type)
  37     , sock_(ctx, type)
  38   {
  39     if(zmq_type_ == ZMQ_SUB) {
  40       sock_.setsockopt(ZMQ_SUBSCRIBE, "", 0) ; // listen to all messages
  41       sock_.connect(endpoint.c_str()) ;
  42     }
  43     else {
  44       sock_.bind(endpoint.c_str());
  45     }
  46   }
  47
  48   bool serveOne(int recv_flags = 0);
  49   void serve() {
  50     while (true) {
  51       serveOne();
  52     }
  53   }
  54
  55   zmq::socket_t& getSocket() {
  56     return sock_;
  57   }
  58
  59  private:
  60   std::shared_ptr<TProcessor> processor_;
  61   int zmq_type_;
  62   zmq::socket_t sock_;
  63 };
  64
  65
  66 class TZmqMultiServer {
  67  public:
  68   void serveOne(long timeout = -1);
  69   void serveForever();
  70
  71   std::vector<TZmqServer*>& servers() {
  72     return servers_;
  73   }
  74
  75  private:
  76   zmq::pollitem_t* setupPoll();
  77   void serveActive(zmq::pollitem_t* items, long timeout);
  78   std::vector<TZmqServer*> servers_;
  79 };
  80
  81
  82 }}} // apache::thrift::server
  83
  84 #endif // #ifndef _THRIFT_SERVER_TZMQSERVER_H_