Robotics Service Bus

Partial connector pooling

fixes #2532

Partial connector pooling

fixes #2532

Spread connection sharing in src/rsb/transport/spread/*.{h,cpp}

fixes #2532

A new class Bus manages one Spread connection and its group
memberships. One or more connectors are attached to the Bus and
notifications are routed between the Spread connection and the
attached connectors:

• Notifications originating from OutConnectors are sent via the Spread
  connection and dispatched to In{Push,Pull}Connectors with matching
  scopes

• Notifications received via the Spread connection are dispatched to
  In{Push,Pull}Connectors with matching scopes

Management and creation of Bus instances is handled by the Factory
class: a Bus instance is created for each Spread daemon to which at
least one connector is connected.

Bus instances are kept alive by shared_ptrs owned by connectors. The
Factory only maintains weak_ptrs to Bus instances. Consequently, Bus
instances are destroyed when all connectors drop their references.

• src/rsb/transport/spread/SpreadConnection.cpp (SpreadConnection::send):
  send messages with SELF_DISCARD since we dispatch to
  In{Push,Pull}Connectors within the process
• src/rsb/transport/spread/Notifications.h: new file; contains classes
  for representing incoming and outgoing notifications
• src/rsb/transport/spread/ReceiverTask.cpp (includes): adapted to new
  notification representation
  (ReceiverTask::Handler::handleIncomingNotification): use new
  notification representation
  (ReceiverTask::execute): likewise
• src/rsb/transport/spread/DeserializingHandler.{h,cpp} (includes):
  adapted to new notification representation
  (DeserializingHandler::handleMessage): call maybeJoinFragments;
  construct IncomingNotification
  (DeserializingHandler::maybeJoinFragments): new method; join
  FragmentedNotification instances into a single Notification instance
• src/rsb/transport/spread/Bus.{h,cpp}: new files; contain Bus class
• src/rsb/transport/spread/ConnectorBase.{h,cpp} (includes): adapted
  for SpreadConnection -> Bus change
  (ConnectorBase::bus): changed connection -> bus
  (ConnectorBase::ConnectorBase): accept Bus instance of
  SpreadConnection
  (ConnectorBase::printContents): adapted accordingly
  (ConnectorBase::getTransportURL): likewise
• src/rsb/transport/spread/InConnector.{h,cpp} (includes): adapted to
  SpreadConnection -> Bus change
  (InConnector): added base class BusSink
  (InConnector::memberships): removed member variable; now handled by
  Bus
  (InConnector::InConnector): accept Bus instead of SpreadConnection
  (InConnector::activate): register with bus instead of activating
  connection
  (InConnector::deactivate): unregister from bus instead of
  deactivating connection
  (InConnector::notificationToEvent): accept reference to notification
  instead of pointer
  (InConnector::handleError): new method; handle Error notification
  from Bus
• src/rsb/transport/spread/InPullConnector.{h,cpp} (includes): adapted
  to SpreadConnection -> Bus change; added
  rsc/threading/SynchronizedQueue.h
  (InPullConnector::InPullConnector): accept Bus instead of
  SpreadConnection
  (InPullConnector::activate): removed method; now handled by base
  class
  (InPullConnector::deactivate): likewise
  (InPullConnector::handleNotification): new method; accept
  notifications from bus
• src/rsb/transport/spread/InPushConnector.{h,cpp} (includes): adapted
  to SpreadConnection -> Bus change; removed ReceiverTask.h
  (InPushConnector::Handler): removed; no longer needed
  (InPushConnector::exec): removed; no longer needed
  (InPushConnector::rec): likewise
  (InPushConnector::handler): likewise
  (InPushConnector::InPushConnector): accept Bus instead of
  SpreadConnection; no need to initialize handler, task or executor
  (InPushConnector::activate): removed method; now handled by base
  class
  (InPushConnector::deactivate): likewise
  (InPushConnector::handleNotification): renamed
  Handle{Incoming -> }Notification; adapted to changed notification
  representation
• src/rsb/transport/spread/OutConnector.{h,cpp} (includes): adapted to
  SpreadConnection -> Bus change;
  (OutConnector::OutConnector): accept Bus instead of SpreadConnection
  (OutConnector::activate): do not activate connection
  (OutConnector::deactivate): do not deactivate connection
  (OutConnector::handle): serialize and fragment event according to
  changed notification representation; then submit to bus for
  distribution
• src/rsb/transport/spread/Factory.{h,cpp} (Factory::buses): new
  member variable; map of Spread daemon names to Bus instances
  (Factory::busesLock): new member variable; lock for Bus map
  (Factory::obtainBus): renamed createConnection -> obtainBus; check
  map of weak pointers to bus instances for an entry that matches the
  requested daemon and is still alive; use that entry or create a new
  Bus instance
  (Factory::createInPushConnector): use obtainBus
  (Factory::createInPullConnector): likewise
  (Factory::createOutConnector): likewise
• test/rsb/transport/spread/SpreadConnectorTest.cpp (header): updated
  copyright
  (createSpreadInPullConnector): create Bus instance, pass to
  connector
  (createSpreadInPushConnector): likewise
  (createSpreadOutConnector): likewise
• src/CMakeLists.txt (SOURCES): added file
  src/rsb/transport/spread/Bus.cpp
  (HEADERS): added file src/rsb/transport/spread/Bus.h

Mention full Spread connection sharing in C++ in news.rst

refs #2532

• news.rst (RSB 0.18): mention full Spread connection sharing in C++
  implementation