Remote Call Framework 3.1
PublishingService.hpp
1 
2 //******************************************************************************
3 // RCF - Remote Call Framework
4 //
5 // Copyright (c) 2005 - 2019, Delta V Software. All rights reserved.
6 // http://www.deltavsoft.com
7 //
8 // RCF is distributed under dual licenses - closed source or GPL.
9 // Consult your particular license for conditions of use.
10 //
11 // If you have not purchased a commercial license, you are using RCF
12 // under GPL terms.
13 //
14 // Version: 3.1
15 // Contact: support <at> deltavsoft.com
16 //
17 //******************************************************************************
18 
19 #ifndef INCLUDE_RCF_PUBLISHINGSERVICE_HPP
20 #define INCLUDE_RCF_PUBLISHINGSERVICE_HPP
21 
22 #include <functional>
23 #include <map>
24 #include <string>
25 
26 #include <RCF/Export.hpp>
27 #include <RCF/PeriodicTimer.hpp>
28 #include <RCF/RcfClient.hpp>
29 #include <RCF/RcfFwd.hpp>
30 #include <RCF/Service.hpp>
31 #include <RCF/ThreadLibrary.hpp>
32 #include <RCF/Timer.hpp>
33 #include <RCF/Tools.hpp>
34 
35 namespace RCF {
36 
38  class RCF_EXPORT PublisherParms
39  {
40  public:
41 
43  void setTopicName(const std::string & topicName);
44 
46  std::string getTopicName() const;
47 
49  void setOnSubscriberConnect(OnSubscriberConnect onSubscriberConnect);
50 
52  void setOnSubscriberDisconnect(OnSubscriberDisconnect onSubscriberDisconnect);
53 
54  private:
55 
56  friend class PublishingService;
57 
58  std::string mTopicName;
59  OnSubscriberConnect mOnSubscriberConnect;
60  OnSubscriberDisconnect mOnSubscriberDisconnect;
61  };
62 
64  class RCF_EXPORT PublisherBase : Noncopyable
65  {
66  public:
67  PublisherBase(PublishingService & pubService, const PublisherParms & parms);
68  ~PublisherBase();
69 
71  std::string getTopicName();
72 
74  std::size_t getSubscriberCount();
75 
77  void close();
78 
79  protected:
80 
81  friend class PublishingService;
82 
83  void init();
84 
85  PublishingService & mPublishingService;
86  PublisherParms mParms;
87  bool mClosed;
88  std::string mTopicName;
89  RcfClientPtr mRcfClientPtr;
90  };
91 
93  template<typename Interface>
94  class Publisher : public PublisherBase
95  {
96  public:
97 
98  typedef typename Interface::RcfClientT RcfClientT;
99 
100  Publisher(PublishingService & pubService, const PublisherParms & parms) :
101  PublisherBase(pubService, parms),
102  mpClient(NULL)
103  {
104  if (mTopicName.empty())
105  {
106  mTopicName = getInterfaceName<Interface>();
107  }
108 
109  mRcfClientPtr.reset( new RcfClientT( ClientStub(mTopicName) ) );
110  mpClient = dynamic_cast<RcfClientT *>(mRcfClientPtr.get());
111 
112  init();
113  }
114 
116  RcfClientT & publish()
117  {
118  RCF_ASSERT(!mClosed);
119  return *mpClient;
120  }
121 
122  private:
123 
124  RcfClientT * mpClient;
125  };
126 
127  class I_RequestSubscription;
128  template<typename T> class RcfClient;
129 
130  class RCF_EXPORT PublishingService :
131  public I_Service,
132  Noncopyable
133  {
134  public:
135  PublishingService();
136 
137  ~PublishingService();
138 
139  template<typename Interface>
140  std::shared_ptr< Publisher<Interface> > createPublisher(
141  const PublisherParms & parms)
142  {
143  std::shared_ptr< Publisher<Interface> > publisherPtr(
144  new Publisher<Interface>(*this, parms) );
145 
146  std::string topicName = publisherPtr->getTopicName();
147 
148  RCF_ASSERT(topicName.size() > 0);
149  Lock lock(mPublishersMutex);
150  mPublishers[topicName] = publisherPtr;
151  return publisherPtr;
152  }
153 
154  void setPingIntervalMs(std::uint32_t pingIntervalMs);
155  std::uint32_t getPingIntervalMs() const;
156 
157 
158  private:
159 
160  friend class RcfClient<I_RequestSubscription>;
161  friend class PublisherBase;
162  friend class PublishingServicePb;
163  friend class RcfSession;
164 
165  std::int32_t RequestSubscription(
166  const std::string &subscriptionName);
167 
168  std::int32_t RequestSubscription(
169  const std::string &subscriptionName,
170  std::uint32_t subToPubPingIntervalMs,
171  std::uint32_t & pubToSubPingIntervalMs);
172 
173  private:
174 
175  void onServiceAdded(RcfServer &server);
176  void onServiceRemoved(RcfServer &server);
177  void onServerStart(RcfServer &server);
178  void onServerStop(RcfServer &server);
179 
180  void addSubscriberTransport(
181  RcfSession &session,
182  const std::string &publisherName,
183  ClientTransportUniquePtrPtr clientTransportUniquePtrPtr);
184 
185  void closePublisher(const std::string & name);
186 
187  // TODO: should be weak_ptr here probably. ~Publisher() will need
188  // to clean them up ?
189  typedef std::map<std::string, PublisherWeakPtr> Publishers;
190 
191  Mutex mPublishersMutex;
192  Publishers mPublishers;
193 
194  std::uint32_t mPingIntervalMs;
195  PeriodicTimer mPeriodicTimer;
196 
197  virtual void onTimer();
198  void pingAllSubscriptions();
199  void harvestExpiredSubscriptions();
200 
201  };
202 
203  typedef std::shared_ptr<PublishingService> PublishingServicePtr;
204 
205 } // namespace RCF
206 
207 #endif // ! INCLUDE_RCF_PUBLISHINGSERVICE_HPP
RcfClientT & publish()
Returns a reference to the RcfClient<> instance to use when publishing messages.
Definition: PublishingService.hpp:116
Represents a server side session, associated with a client connection.
Definition: RcfSession.hpp:67
Controls the client side of a RCF connection.
Definition: ClientStub.hpp:82
std::function< void(RcfSession &, const std::string &)> OnSubscriberDisconnect
Describes a user-provided callback function to be called on the publisher side, whenever a subscriber...
Definition: RcfFwd.hpp:74
Provides RCF server-side functionality.
Definition: RcfServer.hpp:54
Represents a single publisher within a RcfServer. To create a publisher, use RcfServer::createPublish...
Definition: PublishingService.hpp:94
Definition: AmiIoHandler.hpp:24
General configuration of a publisher.
Definition: PublishingService.hpp:38
std::function< bool(RcfSession &, const std::string &)> OnSubscriberConnect
Describes a user-provided callback function to be called on the publisher side, whenever a subscriber...
Definition: RcfFwd.hpp:71
RCF_EXPORT bool init(RcfConfigT *=nullptr)
Reference-counted initialization of RCF library. May be called multiple times (see deinit())...
Base class of all publishers.
Definition: PublishingService.hpp:64