Source · fairmq/zeromq/
TransportFactory.hheader
Described in Transports.
1/********************************************************************************2 * Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *3 * *4 * This software is distributed under the terms of the *5 * GNU Lesser General Public Licence (LGPL) version 3, *6 * copied verbatim in the file "LICENSE" *7 ********************************************************************************/8 9#ifndef FAIR_MQ_ZMQ_TRANSPORTFACTORY_H10#define FAIR_MQ_ZMQ_TRANSPORTFACTORY_H11 12#include <fairmq/zeromq/Context.h>13#include <fairmq/zeromq/Message.h>14#include <fairmq/zeromq/Socket.h>15#include <fairmq/zeromq/Poller.h>16#include <fairmq/zeromq/UnmanagedRegion.h>17#include <fairmq/TransportFactory.h>18#include <fairmq/ProgOptions.h>19 20#include <memory> // unique_ptr, make_unique21#include <string>22#include <vector>23 24namespace fair::mq::zmq25{26 27class TransportFactory final : public fair::mq::TransportFactory28{29 public:30 TransportFactory(const std::string& id = "", const ProgOptions* config = nullptr)31 : fair::mq::TransportFactory(id)32 , fCtx(nullptr)33 {34 int major = 0, minor = 0, patch = 0;35 zmq_version(&major, &minor, &patch);36 LOG(debug) << "Transport: Using ZeroMQ library, version: " << major << "." << minor << "." << patch;37 38 if (config) {39 fCtx = std::make_unique<Context>(config->GetProperty<int>("io-threads", 1));40 } else {41 LOG(debug) << "fair::mq::ProgOptions not available! Using defaults.";42 fCtx = std::make_unique<Context>(1);43 }44 }45 46 TransportFactory(const TransportFactory&) = delete;47 TransportFactory(TransportFactory&&) = delete;48 TransportFactory& operator=(const TransportFactory&) = delete;49 TransportFactory& operator=(TransportFactory&&) = delete;50 51 MessagePtr CreateMessage() override52 {53 return std::make_unique<Message>(this);54 }55 56 MessagePtr CreateMessage(Alignment alignment) override57 {58 return std::make_unique<Message>(alignment, this);59 }60 61 MessagePtr CreateMessage(size_t size) override62 {63 return std::make_unique<Message>(size, this);64 }65 66 MessagePtr CreateMessage(size_t size, Alignment alignment) override67 {68 return std::make_unique<Message>(size, alignment, this);69 }70 71 MessagePtr CreateMessage(void* data, size_t size, fair::mq::FreeFn* ffn, void* hint = nullptr) override72 {73 return std::make_unique<Message>(data, size, ffn, hint, this);74 }75 76 MessagePtr CreateMessage(UnmanagedRegionPtr& region,77 void* data,78 size_t size,79 void* hint = nullptr) override80 {81 return std::make_unique<Message>(region, data, size, hint, this);82 }83 84 SocketPtr CreateSocket(const std::string& type, const std::string& name) override85 {86 return std::make_unique<Socket>(*fCtx, type, name, GetId(), this);87 }88 89 PollerPtr CreatePoller(const std::vector<Channel>& channels) const override90 {91 return std::make_unique<Poller>(channels);92 }93 94 PollerPtr CreatePoller(const std::vector<Channel*>& channels) const override95 {96 return std::make_unique<Poller>(channels);97 }98 99 PollerPtr CreatePoller(const std::unordered_map<std::string, std::vector<Channel>>& channelsMap, const std::vector<std::string>& channelList) const override100 {101 return std::make_unique<Poller>(channelsMap, channelList);102 }103 104 UnmanagedRegionPtr CreateUnmanagedRegion(size_t size, RegionCallback callback, const std::string& path = "", int flags = 0, fair::mq::RegionConfig cfg = fair::mq::RegionConfig()) override105 {106 return CreateUnmanagedRegion(size, 0, callback, nullptr, path, flags, cfg);107 }108 109 UnmanagedRegionPtr CreateUnmanagedRegion(size_t size, RegionBulkCallback bulkCallback, const std::string& path = "", int flags = 0, fair::mq::RegionConfig cfg = fair::mq::RegionConfig()) override110 {111 return CreateUnmanagedRegion(size, 0, nullptr, bulkCallback, path, flags, cfg);112 }113 114 UnmanagedRegionPtr CreateUnmanagedRegion(size_t size, int64_t userFlags, RegionCallback callback, const std::string& path = "", int flags = 0, fair::mq::RegionConfig cfg = fair::mq::RegionConfig()) override115 {116 return CreateUnmanagedRegion(size, userFlags, callback, nullptr, path, flags, cfg);117 }118 119 UnmanagedRegionPtr CreateUnmanagedRegion(size_t size, int64_t userFlags, RegionBulkCallback bulkCallback, const std::string& path = "", int flags = 0, fair::mq::RegionConfig cfg = fair::mq::RegionConfig()) override120 {121 return CreateUnmanagedRegion(size, userFlags, nullptr, bulkCallback, path, flags, cfg);122 }123 124 UnmanagedRegionPtr CreateUnmanagedRegion(size_t size, RegionCallback callback, RegionConfig cfg) override125 {126 return CreateUnmanagedRegion(size, cfg.userFlags, callback, nullptr, cfg.path, cfg.creationFlags, cfg);127 }128 UnmanagedRegionPtr CreateUnmanagedRegion(size_t size, RegionBulkCallback bulkCallback, RegionConfig cfg) override129 {130 return CreateUnmanagedRegion(size, cfg.userFlags, nullptr, bulkCallback, cfg.path, cfg.creationFlags, cfg);131 }132 133 UnmanagedRegionPtr CreateUnmanagedRegion(size_t size, int64_t userFlags, RegionCallback callback, RegionBulkCallback bulkCallback, const std::string&, int /* flags */, fair::mq::RegionConfig cfg)134 {135 UnmanagedRegionPtr ptr = std::make_unique<UnmanagedRegion>(*fCtx, size, userFlags, callback, bulkCallback, this, cfg);136 auto zPtr = static_cast<UnmanagedRegion*>(ptr.get());137 fCtx->AddRegion(false, zPtr->GetId(), zPtr->GetData(), zPtr->GetSize(), zPtr->GetUserFlags(), RegionEvent::created);138 return ptr;139 }140 141 void SubscribeToRegionEvents(RegionEventCallback callback) override { fCtx->SubscribeToRegionEvents(callback); }142 bool SubscribedToRegionEvents() override { return fCtx->SubscribedToRegionEvents(); }143 void UnsubscribeFromRegionEvents() override { fCtx->UnsubscribeFromRegionEvents(); }144 std::vector<RegionInfo> GetRegionInfo() override { return fCtx->GetRegionInfo(); }145 146 Transport GetType() const override { return Transport::ZMQ; }147 148 void Interrupt() override { fCtx->Interrupt(); }149 void Resume() override { fCtx->Resume(); }150 void Reset() override { fCtx->Reset(); }151 152 ~TransportFactory() override { LOG(debug) << "Destroying ZeroMQ transport..."; }153 154 private:155 std::unique_ptr<Context> fCtx;156};157 158} // namespace fair::mq::zmq159 160#endif /* FAIR_MQ_ZMQ_TRANSPORTFACTORY_H */