ark::pipeline::Subscriber

Defined in header “ark/pipeline/subscriber.hh”.

---

This is used to indicate that you are subscribed to a particular piece of data. This subscriber receives data (up to some queue depth) from other publishers in the pipeline, and can invoke a callback when data is ready.

Callback will not be executed simultaneously; in other words, only one instance of a subscriber will be executed at any particular point in time.

Typedefs

• using ConfigurationType = SubscriberConfiguration< Type >
  The configuration type needed.

• using TypePtr = typename ConfigurationType::TypePtr
  The type of data being used.

• using EnvelopePtr = std::conditional_t< is_abstract_envelope, std::shared_ptr< const AbstractEnvelope >, std::shared_ptr< const Envelope< Type > »
  The envelope type that we store/publish.

Methods

• Subscriber(ConfigurationType config)
  Constructor. Initialize a subscriber against the specified channel name.

• SubscriberStatistics statistics()
  Returns the statistics related to this subscriber. These are running statistics, initializing when the subscriber starts, and counting up from there.

• SubscriberTimings timings()
  Returns timing information related to this subscriber. Resets the timing information after this call (so the maximums are reset). This API is meant to be called periodically to determine the max timing information over some period of time.

• void push(DataEnvelopePtr message)
  Enqueues the given data envelope onto the subscriber. The object will be deserialized into a concrete type, and then queued for execution as if you had called push() on the concrete type.

• void push_data(DataEnvelopeCPtr message)
  Enqueues the given data envelope onto the subscriber. The object will be deserialized into a concrete type, and then queued for execution as if you had called push() on the concrete type.

• void push(AbstractEnvelopeCPtr message)
  Enqueues the given envelope onto the subscriber. If the queue exceeds it’s limits, the oldest item on the queue will be dropped.

  If you enqueue an envelope that does not match the subscribers type, this will throw an error.

• void push_concrete(EnvelopePtr message)
  Enqueues the given envelope onto the subscriber. If the queue exceeds it’s limits, the oldest item on the queue will be dropped.

  This assumes the type already matches the type that this producer publishes.

• const std::type_info & message_type()
  Returns type information about the messages that flow through this publisher.

• bool consumes_type(const std::type_info & type)
  Returns true if this subscriber can execute messages of the given type or not.

• bool consumes_message_type(const std::string & type)
  Returns true if this subscriber can execute messages of the given message type or not. Assumes the type is serializable.

• bool message_is_serializable()
  Returns true if the message type for this subscriber is serializable.

• std::string message_schema_protocol()
  Returns the message schema protocol (rbuf, proto, etc) of the object that flows over this subscriber.

• std::string message_type_name()
  Returns the message type name of the object that flows over this subscriber.

• std::string message_schema_registry()
  Returns the schema for the object that flows over this subscriber, if available.

• size_t queue_size()
  Returns the size of the queue for this subscriber. This value might change frequently; this is just a snapshot.

• bool pop(SubscriberWorkItem & item)
  Pops a work item off the subscriber. The execute() function is actually broken into two halves, a phase for popping off the queue, and a phase for running the callback. This is the first half, allowing for you to deterministically pop an envelope off a subscriber.

  @oaram[in] item Work item to populate

• void execute_work_item(SubscriberWorkItem && item)
  Executes the given work item, as received from a call to pop(). Consumes the work item, taking ownership of it.

• bool execute()
  Pops the oldest item off the queue and executes the callback against it. Returns false if the queue is empty.

• size_t configured_max_queue_depth()
  Returns the max queue depth configured for this subscriber.

• std::chrono::nanoseconds expected_runtime(const SubscriberWorkItem & work_item)
  Returns the expected runtime as if this subscriber had executed the given work item.

• std::chrono::nanoseconds configured_expected_runtime()
  Returns the configured execution time for this subscriber.

• void record_publisher_connection()
  Marks that this subscriber has been connected to a new publisher.