src/thread/disruptor/interface.h - rocketmq-client-cpp - Git at Google

 // Copyright (c) 2011, François Saint-Jacques
 // All rights reserved.
 //
 // Redistribution and use in source and binary forms, with or without
 // modification, are permitted provided that the following conditions are met:
 //     * Redistributions of source code must retain the above copyright
 //       notice, this list of conditions and the following disclaimer.
 //     * Redistributions in binary form must reproduce the above copyright
 //       notice, this list of conditions and the following disclaimer in the
 //       documentation and/or other materials provided with the distribution.
 //     * Neither the name of the disruptor-- nor the
 //       names of its contributors may be used to endorse or promote products
 //       derived from this software without specific prior written permission.
 //
 // THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
 // ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
 // WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 // DISCLAIMED. IN NO EVENT SHALL FRANÇOIS SAINT-JACQUES BE LIABLE FOR ANY
 // DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 // (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 // LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
 // ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 // (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 // SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

 #ifndef DISRUPTOR_INTERFACE_H_ // NOLINT
 #define DISRUPTOR_INTERFACE_H_ // NOLINT

 #include <climits>
 #include <vector>

 #include "sequence.h"
 #include "batch_descriptor.h"

 namespace rocketmq {

 // Strategies employed for claiming the sequence of events in the
 // {@link Seqencer} by publishers.
 class ClaimStrategyInterface {
  public:
     // Is there available capacity in the buffer for the requested sequence.
     //
     // @param dependent_sequences to be checked for range.
     // @return true if the buffer has capacity for the requested sequence.
     virtual ~ClaimStrategyInterface() {}
     virtual bool HasAvalaibleCapacity(
         const std::vector<Sequence*>& dependent_sequences) = 0;

     // Claim the next sequence in the {@link Sequencer}.
     //
     // @param dependent_sequences to be checked for range.
     // @return the index to be used for the publishing.
     virtual int64_t IncrementAndGet(
             const std::vector<Sequence*>& dependent_sequences) = 0;

     // Claim the next sequence in the {@link Sequencer}.
     //
     // @param delta to increment by.
     // @param dependent_sequences to be checked for range.
     // @return the index to be used for the publishing.
     virtual int64_t IncrementAndGet(const int& delta,
             const std::vector<Sequence*>& dependent_sequences) = 0;

     // Set the current sequence value for claiming an event in the
     // {@link Sequencer}.
     //
     // @param sequence to be set as the current value.
     // @param dependent_sequences to be checked for range.
     virtual void SetSequence(const int64_t& sequence,
             const std::vector<Sequence*>& dependent_sequences) = 0;

     // Serialise publishing in sequence.
     //
     // @param sequence to be applied.
     // @param cursor to be serialise against.
     // @param batch_size of the sequence.
     virtual void SerialisePublishing(const int64_t& sequence,
                                      const Sequence& cursor,
                                      const int64_t& batch_size) = 0;
 };

 // Coordination barrier for tracking the cursor for publishers and sequence of
 // dependent {@link EventProcessor}s for processing a data structure.
 class SequenceBarrierInterface {
  public:
     // Wait for the given sequence to be available for consumption.
     //
     // @param sequence to wait for.
     // @return the sequence up to which is available.
     //
     // @throws AlertException if a status change has occurred for the
     // Disruptor.
     virtual ~SequenceBarrierInterface(){}
     virtual int64_t WaitFor(const int64_t& sequence) = 0;

     // Wait for the given sequence to be available for consumption with a
     // time out.
     //
     // @param sequence to wait for.
     // @param timeout in microseconds.
     // @return the sequence up to which is available.
     //
     // @throws AlertException if a status change has occurred for the
     // Disruptor.
     virtual int64_t WaitFor(const int64_t& sequence,
                             const int64_t& timeout_micro) = 0;

     // Delegate a call to the {@link Sequencer#getCursor()}
     //
     //  @return value of the cursor for entries that have been published.
     virtual int64_t GetCursor() const = 0;

     // The current alert status for the barrier.
     //
     // @return true if in alert otherwise false.
     virtual bool IsAlerted() const = 0;

     // Alert the {@link EventProcessor}s of a status change and stay in this
     // status until cleared.
     virtual void Alert() = 0;

     // Clear the current alert status.
     virtual void ClearAlert() = 0;

     // Check if barrier is alerted, if so throws an AlertException
     //
     // @throws AlertException if barrier is alerted
     virtual void CheckAlert() const = 0;
 };

 // Called by the {@link RingBuffer} to pre-populate all the events to fill the
 // RingBuffer.
 //
 // @param <T> event implementation storing the data for sharing during exchange
 // or parallel coordination of an event.
 template<typename T>
 class EventFactoryInterface {
  public:
     virtual ~EventFactoryInterface(){}
      virtual T* NewInstance(const int& size) const = 0;
 };

 // Callback interface to be implemented for processing events as they become
 // available in the {@link RingBuffer}.
 //
 // @param <T> event implementation storing the data for sharing during exchange
 // or parallel coordination of an event.
 template<typename T>
 class EventHandlerInterface {
  public:
     // Called when a publisher has published an event to the {@link RingBuffer}
     //
     // @param event published to the {@link RingBuffer}
     // @param sequence of the event being processed
     // @param end_of_batch flag to indicate if this is the last event in a batch
     // from the {@link RingBuffer}
     //
     // @throws Exception if the EventHandler would like the exception handled
     // further up the chain.
     virtual ~EventHandlerInterface(){}
     virtual void OnEvent(const int64_t& sequence,
                          const bool& end_of_batch,
                          T* event)  = 0;

     // Called once on thread start before processing the first event.
     virtual void OnStart() = 0;

     // Called once on thread stop just before shutdown.
     virtual void OnShutdown() = 0;
 };

 // Implementations translate another data representations into events claimed
 // for the {@link RingBuffer}.
 //
 // @param <T> event implementation storing the data for sharing during exchange
 // or parallel coordination of an event.
 template<typename T>
 class EventTranslatorInterface {
  public:
      // Translate a data representation into fields set in given event
      //
      // @param event into which the data should be translated.
      // @param sequence that is assigned to events.
      // @return the resulting event after it has been translated.
      virtual ~EventTranslatorInterface(){}
      virtual T* TranslateTo(const int64_t& sequence, T* event) { return NULL;}
 };

 // EventProcessors wait for events to become available for consumption from
 // the {@link RingBuffer}. An event processor should be associated with a
 // thread.
 //
 // @param <T> event implementation storing the data for sharing during exchange
 // or parallel coordination of an event.
 template<typename T>
 class EventProcessorInterface {
  public:
      // Get a pointer to the {@link Sequence} being used by this
      // {@link EventProcessor}.
      //
      // @return pointer to the {@link Sequence} for this
      // {@link EventProcessor}
      virtual ~EventProcessorInterface(){}
     virtual Sequence* GetSequence() = 0;

     // Signal that this EventProcessor should stop when it has finished
     // consuming at the next clean break.
     // It will call {@link DependencyBarrier#alert()} to notify the thread to
     // check status.
     virtual void Halt() = 0;
 };

 // Callback handler for uncaught exception in the event processing cycle
 // of the {@link BatchEventProcessor}.
 //
 // @param <T> event type stored in the {@link RingBuffer}.
 template<typename T>
 class ExceptionHandlerInterface {
  public:
     // Strategy for handling uncaught exceptions when processing an event.
     // If the strategy wishes to suspend further processing by the
     // {@link BatchEventProcessor} then it should throw a std::runtime_error.
     //
     // @param exception that propagated from the {@link EventHandler}.
     // @param sequence of the event which caused the exception.
     // @param event being processed when the exception occured.
     virtual ~ExceptionHandlerInterface(){}
     virtual void Handle(const std::exception& exception,
                         const int64_t& sequence,
                         T* event) = 0;
 };

 // Strategy employed for making {@link EventProcessor}s wait on a cursor
 // {@link Sequence}.
 class WaitStrategyInterface: public boost::noncopyable {
  public:
     //  Wait for the given sequence to be available for consumption.
     //
     //  @param dependents further back the chain that must advance first.
     //  @param cursor on which to wait.
     //  @param barrier the consumer is waiting on.
     //  @param sequence to be waited on.
     //  @return the sequence that is available which may be greater than the
     //  requested sequence.
     //
     //  @throws AlertException if the status of the Disruptor has changed.
     virtual ~WaitStrategyInterface(){}
     virtual int64_t WaitFor(const std::vector<Sequence*>& dependents,
                             const Sequence& cursor,
                             const SequenceBarrierInterface& barrier,
                             const int64_t& sequence) = 0;

     //  Wait for the given sequence to be available for consumption in a
     //  {@link RingBuffer} with a timeout specified.
     //
     //  @param dependents further back the chain that must advance first
     //  @param cursor on which to wait.
     //  @param barrier the consumer is waiting on.
     //  @param sequence to be waited on.
     //  @param timeout value in micro seconds to abort after.
     //  @return the sequence that is available which may be greater than the
     //  requested sequence.
     //
     //  @throws AlertException if the status of the Disruptor has changed.
     //  @throws InterruptedException if the thread is interrupted.
     virtual int64_t WaitFor(const std::vector<Sequence*>& dependents,
                             const Sequence& cursor,
                             const SequenceBarrierInterface& barrier,
                             const int64_t & sequence,
                             const int64_t & timeout_micros) = 0;

     // Signal those waiting that the cursor has advanced.
     virtual void SignalAllWhenBlocking() = 0;
 };

 };  // namespace rocketmq

 #endif // DISRUPTOR_INTERFACE_H_ NOLINT
	// Copyright (c) 2011, François Saint-Jacques
	// All rights reserved.
	//
	// Redistribution and use in source and binary forms, with or without
	// modification, are permitted provided that the following conditions are met:
	// * Redistributions of source code must retain the above copyright
	// notice, this list of conditions and the following disclaimer.
	// * Redistributions in binary form must reproduce the above copyright
	// notice, this list of conditions and the following disclaimer in the
	// documentation and/or other materials provided with the distribution.
	// * Neither the name of the disruptor-- nor the
	// names of its contributors may be used to endorse or promote products
	// derived from this software without specific prior written permission.
	//
	// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
	// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
	// WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
	// DISCLAIMED. IN NO EVENT SHALL FRANÇOIS SAINT-JACQUES BE LIABLE FOR ANY
	// DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
	// (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
	// LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
	// ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
	// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
	// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

	#ifndef DISRUPTOR_INTERFACE_H_ // NOLINT
	#define DISRUPTOR_INTERFACE_H_ // NOLINT

	#include <climits>
	#include <vector>

	#include "sequence.h"
	#include "batch_descriptor.h"

	namespace rocketmq {

	// Strategies employed for claiming the sequence of events in the
	// {@link Seqencer} by publishers.
	class ClaimStrategyInterface {
	public:
	// Is there available capacity in the buffer for the requested sequence.
	//
	// @param dependent_sequences to be checked for range.
	// @return true if the buffer has capacity for the requested sequence.
	virtual ~ClaimStrategyInterface() {}
	virtual bool HasAvalaibleCapacity(
	const std::vector<Sequence*>& dependent_sequences) = 0;

	// Claim the next sequence in the {@link Sequencer}.
	//
	// @param dependent_sequences to be checked for range.
	// @return the index to be used for the publishing.
	virtual int64_t IncrementAndGet(
	const std::vector<Sequence*>& dependent_sequences) = 0;

	// Claim the next sequence in the {@link Sequencer}.
	//
	// @param delta to increment by.
	// @param dependent_sequences to be checked for range.
	// @return the index to be used for the publishing.
	virtual int64_t IncrementAndGet(const int& delta,
	const std::vector<Sequence*>& dependent_sequences) = 0;

	// Set the current sequence value for claiming an event in the
	// {@link Sequencer}.
	//
	// @param sequence to be set as the current value.
	// @param dependent_sequences to be checked for range.
	virtual void SetSequence(const int64_t& sequence,
	const std::vector<Sequence*>& dependent_sequences) = 0;

	// Serialise publishing in sequence.
	//
	// @param sequence to be applied.
	// @param cursor to be serialise against.
	// @param batch_size of the sequence.
	virtual void SerialisePublishing(const int64_t& sequence,
	const Sequence& cursor,
	const int64_t& batch_size) = 0;
	};

	// Coordination barrier for tracking the cursor for publishers and sequence of
	// dependent {@link EventProcessor}s for processing a data structure.
	class SequenceBarrierInterface {
	public:
	// Wait for the given sequence to be available for consumption.
	//
	// @param sequence to wait for.
	// @return the sequence up to which is available.
	//
	// @throws AlertException if a status change has occurred for the
	// Disruptor.
	virtual ~SequenceBarrierInterface(){}
	virtual int64_t WaitFor(const int64_t& sequence) = 0;

	// Wait for the given sequence to be available for consumption with a
	// time out.
	//
	// @param sequence to wait for.
	// @param timeout in microseconds.
	// @return the sequence up to which is available.
	//
	// @throws AlertException if a status change has occurred for the
	// Disruptor.
	virtual int64_t WaitFor(const int64_t& sequence,
	const int64_t& timeout_micro) = 0;

	// Delegate a call to the {@link Sequencer#getCursor()}
	//
	// @return value of the cursor for entries that have been published.
	virtual int64_t GetCursor() const = 0;

	// The current alert status for the barrier.
	//
	// @return true if in alert otherwise false.
	virtual bool IsAlerted() const = 0;

	// Alert the {@link EventProcessor}s of a status change and stay in this
	// status until cleared.
	virtual void Alert() = 0;

	// Clear the current alert status.
	virtual void ClearAlert() = 0;

	// Check if barrier is alerted, if so throws an AlertException
	//
	// @throws AlertException if barrier is alerted
	virtual void CheckAlert() const = 0;
	};

	// Called by the {@link RingBuffer} to pre-populate all the events to fill the
	// RingBuffer.
	//
	// @param <T> event implementation storing the data for sharing during exchange
	// or parallel coordination of an event.
	template<typename T>
	class EventFactoryInterface {
	public:
	virtual ~EventFactoryInterface(){}
	virtual T* NewInstance(const int& size) const = 0;
	};

	// Callback interface to be implemented for processing events as they become
	// available in the {@link RingBuffer}.
	//
	// @param <T> event implementation storing the data for sharing during exchange
	// or parallel coordination of an event.
	template<typename T>
	class EventHandlerInterface {
	public:
	// Called when a publisher has published an event to the {@link RingBuffer}
	//
	// @param event published to the {@link RingBuffer}
	// @param sequence of the event being processed
	// @param end_of_batch flag to indicate if this is the last event in a batch
	// from the {@link RingBuffer}
	//
	// @throws Exception if the EventHandler would like the exception handled
	// further up the chain.
	virtual ~EventHandlerInterface(){}
	virtual void OnEvent(const int64_t& sequence,
	const bool& end_of_batch,
	T* event) = 0;

	// Called once on thread start before processing the first event.
	virtual void OnStart() = 0;

	// Called once on thread stop just before shutdown.
	virtual void OnShutdown() = 0;
	};

	// Implementations translate another data representations into events claimed
	// for the {@link RingBuffer}.
	//
	// @param <T> event implementation storing the data for sharing during exchange
	// or parallel coordination of an event.
	template<typename T>
	class EventTranslatorInterface {
	public:
	// Translate a data representation into fields set in given event
	//
	// @param event into which the data should be translated.
	// @param sequence that is assigned to events.
	// @return the resulting event after it has been translated.
	virtual ~EventTranslatorInterface(){}
	virtual T* TranslateTo(const int64_t& sequence, T* event) { return NULL;}
	};

	// EventProcessors wait for events to become available for consumption from
	// the {@link RingBuffer}. An event processor should be associated with a
	// thread.
	//
	// @param <T> event implementation storing the data for sharing during exchange
	// or parallel coordination of an event.
	template<typename T>
	class EventProcessorInterface {
	public:
	// Get a pointer to the {@link Sequence} being used by this
	// {@link EventProcessor}.
	//
	// @return pointer to the {@link Sequence} for this
	// {@link EventProcessor}
	virtual ~EventProcessorInterface(){}
	virtual Sequence* GetSequence() = 0;

	// Signal that this EventProcessor should stop when it has finished
	// consuming at the next clean break.
	// It will call {@link DependencyBarrier#alert()} to notify the thread to
	// check status.
	virtual void Halt() = 0;
	};

	// Callback handler for uncaught exception in the event processing cycle
	// of the {@link BatchEventProcessor}.
	//
	// @param <T> event type stored in the {@link RingBuffer}.
	template<typename T>
	class ExceptionHandlerInterface {
	public:
	// Strategy for handling uncaught exceptions when processing an event.
	// If the strategy wishes to suspend further processing by the
	// {@link BatchEventProcessor} then it should throw a std::runtime_error.
	//
	// @param exception that propagated from the {@link EventHandler}.
	// @param sequence of the event which caused the exception.
	// @param event being processed when the exception occured.
	virtual ~ExceptionHandlerInterface(){}
	virtual void Handle(const std::exception& exception,
	const int64_t& sequence,
	T* event) = 0;
	};

	// Strategy employed for making {@link EventProcessor}s wait on a cursor
	// {@link Sequence}.
	class WaitStrategyInterface: public boost::noncopyable {
	public:
	// Wait for the given sequence to be available for consumption.
	//
	// @param dependents further back the chain that must advance first.
	// @param cursor on which to wait.
	// @param barrier the consumer is waiting on.
	// @param sequence to be waited on.
	// @return the sequence that is available which may be greater than the
	// requested sequence.
	//
	// @throws AlertException if the status of the Disruptor has changed.
	virtual ~WaitStrategyInterface(){}
	virtual int64_t WaitFor(const std::vector<Sequence*>& dependents,
	const Sequence& cursor,
	const SequenceBarrierInterface& barrier,
	const int64_t& sequence) = 0;

	// Wait for the given sequence to be available for consumption in a
	// {@link RingBuffer} with a timeout specified.
	//
	// @param dependents further back the chain that must advance first
	// @param cursor on which to wait.
	// @param barrier the consumer is waiting on.
	// @param sequence to be waited on.
	// @param timeout value in micro seconds to abort after.
	// @return the sequence that is available which may be greater than the
	// requested sequence.
	//
	// @throws AlertException if the status of the Disruptor has changed.
	// @throws InterruptedException if the thread is interrupted.
	virtual int64_t WaitFor(const std::vector<Sequence*>& dependents,
	const Sequence& cursor,
	const SequenceBarrierInterface& barrier,
	const int64_t & sequence,
	const int64_t & timeout_micros) = 0;

	// Signal those waiting that the cursor has advanced.
	virtual void SignalAllWhenBlocking() = 0;
	};

	}; // namespace rocketmq

	#endif // DISRUPTOR_INTERFACE_H_ NOLINT