components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/DisruptorEndpoint.java - camel - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
  * the License.  You may obtain a copy of the License at
  *
  *      http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package org.apache.camel.component.disruptor;

 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArraySet;

 import com.lmax.disruptor.InsufficientCapacityException;
 import org.apache.camel.Component;
 import org.apache.camel.Consumer;
 import org.apache.camel.Exchange;
 import org.apache.camel.MultipleConsumersSupport;
 import org.apache.camel.Processor;
 import org.apache.camel.Producer;
 import org.apache.camel.WaitForTaskToComplete;
 import org.apache.camel.api.management.ManagedAttribute;
 import org.apache.camel.api.management.ManagedResource;
 import org.apache.camel.impl.DefaultEndpoint;
 import org.apache.camel.spi.Metadata;
 import org.apache.camel.spi.UriEndpoint;
 import org.apache.camel.spi.UriParam;
 import org.apache.camel.spi.UriPath;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 /**
  * The disruptor component provides asynchronous SEDA behavior using LMAX Disruptor.
  *
  * This component works much as the standard SEDA Component, but utilizes a Disruptor
  * instead of a BlockingQueue utilized by the standard SEDA.
  */
 @ManagedResource(description = "Managed Disruptor Endpoint")
 @UriEndpoint(scheme = "disruptor,disruptor-vm", title = "Disruptor,Disruptor VM", syntax = "disruptor:name", consumerClass = DisruptorConsumer.class, label = "endpoint")
 public class DisruptorEndpoint extends DefaultEndpoint implements MultipleConsumersSupport {
     public static final String DISRUPTOR_IGNORE_EXCHANGE = "disruptor.ignoreExchange";
     private static final Logger LOGGER = LoggerFactory.getLogger(DisruptorEndpoint.class);

     private final Set<DisruptorProducer> producers = new CopyOnWriteArraySet<DisruptorProducer>();
     private final Set<DisruptorConsumer> consumers = new CopyOnWriteArraySet<DisruptorConsumer>();
     private final DisruptorReference disruptorReference;

     @UriPath(description = "Name of queue") @Metadata(required = "true")
     private String name;
     @UriParam(label = "consumer", defaultValue = "1")
     private final int concurrentConsumers;
     @UriParam(label = "consumer")
     private final boolean multipleConsumers;
     @UriParam(label = "producer", defaultValue = "IfReplyExpected")
     private WaitForTaskToComplete waitForTaskToComplete = WaitForTaskToComplete.IfReplyExpected;
     @UriParam(label = "producer", defaultValue = "30000")
     private long timeout = 30000;
     @UriParam(defaultValue = "" + DisruptorComponent.DEFAULT_BUFFER_SIZE)
     private int size;
     @UriParam(label = "producer")
     private boolean blockWhenFull;
     @UriParam(label = "consumer", defaultValue = "Blocking")
     private DisruptorWaitStrategy waitStrategy;
     @UriParam(label = "producer", defaultValue = "Multi")
     private DisruptorProducerType producerType;

     public DisruptorEndpoint(final String endpointUri, final Component component,
                              final DisruptorReference disruptorReference, final int concurrentConsumers,
                              final boolean multipleConsumers, boolean blockWhenFull) throws Exception {
         super(endpointUri, component);
         this.disruptorReference = disruptorReference;
         this.name = disruptorReference.getName();
         this.concurrentConsumers = concurrentConsumers;
         this.multipleConsumers = multipleConsumers;
         this.blockWhenFull = blockWhenFull;
     }

     @ManagedAttribute(description = "Queue name")
     public String getName() {
         return name;
     }

     @ManagedAttribute(description = "Buffer max capacity")
     public int getBufferSize() {
         return disruptorReference.getBufferSize();
     }

     @ManagedAttribute(description = "Remaining capacity in ring buffer")
     public long getRemainingCapacity() throws DisruptorNotStartedException {
         return getDisruptor().getRemainingCapacity();
     }

     @ManagedAttribute(description = "Amount of pending exchanges waiting for consumption in ring buffer")
     public long getPendingExchangeCount() throws DisruptorNotStartedException {
         return getDisruptor().getPendingExchangeCount();
     }

     /**
      * Number of concurrent threads processing exchanges.
      */
     @ManagedAttribute(description = "Number of concurrent consumers")
     public int getConcurrentConsumers() {
         return concurrentConsumers;
     }

     @ManagedAttribute(description = "Option to specify whether the caller should wait for the async task to complete or not before continuing")
     public WaitForTaskToComplete getWaitForTaskToComplete() {
         return waitForTaskToComplete;
     }

     /**
      * Option to specify whether the caller should wait for the async task to complete or not before continuing.
      * The following three options are supported: Always, Never or IfReplyExpected. The first two values are self-explanatory.
      * The last value, IfReplyExpected, will only wait if the message is Request Reply based.
      */
     public void setWaitForTaskToComplete(final WaitForTaskToComplete waitForTaskToComplete) {
         this.waitForTaskToComplete = waitForTaskToComplete;
     }

     @ManagedAttribute(description = "Timeout (in milliseconds) before a producer will stop waiting for an asynchronous task to complete")
     public long getTimeout() {
         return timeout;
     }

     /**
      * Timeout (in milliseconds) before a producer will stop waiting for an asynchronous task to complete.
      * You can disable timeout by using 0 or a negative value.
      */
     public void setTimeout(final long timeout) {
         this.timeout = timeout;
     }

     @ManagedAttribute(description = "The maximum capacity of the Disruptors ringbuffer")
     public int getSize() {
         return size;
     }

     /**
      * The maximum capacity of the Disruptors ringbuffer
      * Will be effectively increased to the nearest power of two.
      * Notice: Mind if you use this option, then its the first endpoint being created with the queue name,
      * that determines the size. To make sure all endpoints use same size, then configure the size option
      * on all of them, or the first endpoint being created.
      */
     public void setSize(int size) {
         this.size = size;
     }

     @Override
     @ManagedAttribute(description = "Specifies whether multiple consumers are allowed")
     public boolean isMultipleConsumersSupported() {
         return isMultipleConsumers();
     }

     /**
      * Specifies whether multiple consumers are allowed.
      * If enabled, you can use Disruptor for Publish-Subscribe messaging.
      * That is, you can send a message to the queue and have each consumer receive a copy of the message.
      * When enabled, this option should be specified on every consumer endpoint.
      */
     public boolean isMultipleConsumers() {
         return multipleConsumers;
     }

     /**
      * Returns the current active consumers on this endpoint
      */
     public Set<DisruptorConsumer> getConsumers() {
         return Collections.unmodifiableSet(consumers);
     }

     /**
      * Returns the current active producers on this endpoint
      */
     public Set<DisruptorProducer> getProducers() {
         return Collections.unmodifiableSet(producers);
     }

     @ManagedAttribute
     public boolean isBlockWhenFull() {
         return blockWhenFull;
     }

     /**
      * Whether a thread that sends messages to a full Disruptor will block until the ringbuffer's capacity is no longer exhausted.
      * By default, the calling thread will block and wait until the message can be accepted.
      * By disabling this option, an exception will be thrown stating that the queue is full.
      */
     public void setBlockWhenFull(boolean blockWhenFull) {
         this.blockWhenFull = blockWhenFull;
     }

     @ManagedAttribute(description = "Defines the strategy used by consumer threads to wait on new exchanges to be published")
     public DisruptorWaitStrategy getWaitStrategy() {
         return waitStrategy;
     }

     /**
      * Defines the strategy used by consumer threads to wait on new exchanges to be published.
      * The options allowed are:Blocking, Sleeping, BusySpin and Yielding.
      */
     public void setWaitStrategy(DisruptorWaitStrategy waitStrategy) {
         this.waitStrategy = waitStrategy;
     }

     @ManagedAttribute(description = " Defines the producers allowed on the Disruptor")
     public DisruptorProducerType getProducerType() {
         return producerType;
     }

     /**
      * Defines the producers allowed on the Disruptor.
      * The options allowed are: Multi to allow multiple producers and Single to enable certain optimizations only
      * allowed when one concurrent producer (on one thread or otherwise synchronized) is active.
      */
     public void setProducerType(DisruptorProducerType producerType) {
         this.producerType = producerType;
     }

     @Override
     public boolean isSingleton() {
         return true;
     }

     @Override
     public Producer createProducer() throws Exception {
         if (getProducers().size() == 1 && getDisruptor().getProducerType() == DisruptorProducerType.Single) {
             throw new IllegalStateException(
                     "Endpoint can't support multiple producers when ProducerType SINGLE is configured");
         }
         return new DisruptorProducer(this, getWaitForTaskToComplete(), getTimeout(), isBlockWhenFull());
     }

     @Override
     public Consumer createConsumer(final Processor processor) throws Exception {
         return new DisruptorConsumer(this, processor);
     }

     @Override
     protected void doStart() throws Exception {
         // notify reference we are shutting down this endpoint
         disruptorReference.addEndpoint(this);
         super.doStart();
     }

     @Override
     protected void doStop() throws Exception {
         // notify reference we are shutting down this endpoint
         disruptorReference.removeEndpoint(this);
         super.doStop();
     }

     @Override
     protected void doShutdown() throws Exception {
         // notify component we are shutting down this endpoint
         if (getComponent() != null) {
             getComponent().onShutdownEndpoint(this);
         }

         super.doShutdown();
     }

     @Override
     public DisruptorComponent getComponent() {
         return (DisruptorComponent)super.getComponent();
     }

     void onStarted(final DisruptorConsumer consumer) throws Exception {
         synchronized (this) {
             // validate multiple consumers has been enabled is necessary
             if (!consumers.isEmpty() && !isMultipleConsumersSupported()) {
                 throw new IllegalStateException(
                         "Multiple consumers for the same endpoint is not allowed: " + this);
             }
             if (consumers.add(consumer)) {
                 LOGGER.debug("Starting consumer {} on endpoint {}", consumer, getEndpointUri());
                 getDisruptor().reconfigure();
             } else {
                 LOGGER.debug("Tried to start Consumer {} on endpoint {} but it was already started", consumer, getEndpointUri());
             }
         }
     }


     void onStopped(final DisruptorConsumer consumer) throws Exception {
         synchronized (this) {
             if (consumers.remove(consumer)) {
                 LOGGER.debug("Stopping consumer {} on endpoint {}", consumer, getEndpointUri());
                 getDisruptor().reconfigure();
             } else {
                 LOGGER.debug("Tried to stop Consumer {} on endpoint {} but it was already stopped", consumer, getEndpointUri());
             }
         }
     }

     void onStarted(final DisruptorProducer producer) {
         producers.add(producer);
     }

     void onStopped(final DisruptorProducer producer) {
         producers.remove(producer);
     }

     Map<DisruptorConsumer, Collection<LifecycleAwareExchangeEventHandler>> createConsumerEventHandlers() {
         Map<DisruptorConsumer, Collection<LifecycleAwareExchangeEventHandler>> result =
                 new HashMap<DisruptorConsumer, Collection<LifecycleAwareExchangeEventHandler>>();

         for (final DisruptorConsumer consumer : consumers) {
             result.put(consumer, consumer.createEventHandlers(concurrentConsumers));
         }

         return result;
     }

     /**
      * Called by DisruptorProducers to publish new exchanges on the RingBuffer, blocking when full
      */
     void publish(final Exchange exchange) throws DisruptorNotStartedException {
         disruptorReference.publish(exchange);
     }

     /**
      * Called by DisruptorProducers to publish new exchanges on the RingBuffer, throwing InsufficientCapacityException
      * when full
      *
      * @throws InsufficientCapacityException when the Ringbuffer is full.
      */
     void tryPublish(final Exchange exchange) throws DisruptorNotStartedException, InsufficientCapacityException {
         disruptorReference.tryPublish(exchange);
     }

     DisruptorReference getDisruptor() {
         return disruptorReference;
     }

     @Override
     public boolean equals(Object object) {
         boolean result = super.equals(object);
         return result && getCamelContext().equals(((DisruptorEndpoint)object).getCamelContext());
     }

     @Override
     public int hashCode() {
         return getEndpointUri().hashCode() * 37 + getCamelContext().hashCode();
     }
 }
	/**
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You under the Apache License, Version 2.0
	* (the "License"); you may not use this file except in compliance with
	* the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	package org.apache.camel.component.disruptor;

	import java.util.Collection;
	import java.util.Collections;
	import java.util.HashMap;
	import java.util.Map;
	import java.util.Set;
	import java.util.concurrent.CopyOnWriteArraySet;

	import com.lmax.disruptor.InsufficientCapacityException;
	import org.apache.camel.Component;
	import org.apache.camel.Consumer;
	import org.apache.camel.Exchange;
	import org.apache.camel.MultipleConsumersSupport;
	import org.apache.camel.Processor;
	import org.apache.camel.Producer;
	import org.apache.camel.WaitForTaskToComplete;
	import org.apache.camel.api.management.ManagedAttribute;
	import org.apache.camel.api.management.ManagedResource;
	import org.apache.camel.impl.DefaultEndpoint;
	import org.apache.camel.spi.Metadata;
	import org.apache.camel.spi.UriEndpoint;
	import org.apache.camel.spi.UriParam;
	import org.apache.camel.spi.UriPath;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	/**
	* The disruptor component provides asynchronous SEDA behavior using LMAX Disruptor.
	*
	* This component works much as the standard SEDA Component, but utilizes a Disruptor
	* instead of a BlockingQueue utilized by the standard SEDA.
	*/
	@ManagedResource(description = "Managed Disruptor Endpoint")
	@UriEndpoint(scheme = "disruptor,disruptor-vm", title = "Disruptor,Disruptor VM", syntax = "disruptor:name", consumerClass = DisruptorConsumer.class, label = "endpoint")
	public class DisruptorEndpoint extends DefaultEndpoint implements MultipleConsumersSupport {
	public static final String DISRUPTOR_IGNORE_EXCHANGE = "disruptor.ignoreExchange";
	private static final Logger LOGGER = LoggerFactory.getLogger(DisruptorEndpoint.class);

	private final Set<DisruptorProducer> producers = new CopyOnWriteArraySet<DisruptorProducer>();
	private final Set<DisruptorConsumer> consumers = new CopyOnWriteArraySet<DisruptorConsumer>();
	private final DisruptorReference disruptorReference;

	@UriPath(description = "Name of queue") @Metadata(required = "true")
	private String name;
	@UriParam(label = "consumer", defaultValue = "1")
	private final int concurrentConsumers;
	@UriParam(label = "consumer")
	private final boolean multipleConsumers;
	@UriParam(label = "producer", defaultValue = "IfReplyExpected")
	private WaitForTaskToComplete waitForTaskToComplete = WaitForTaskToComplete.IfReplyExpected;
	@UriParam(label = "producer", defaultValue = "30000")
	private long timeout = 30000;
	@UriParam(defaultValue = "" + DisruptorComponent.DEFAULT_BUFFER_SIZE)
	private int size;
	@UriParam(label = "producer")
	private boolean blockWhenFull;
	@UriParam(label = "consumer", defaultValue = "Blocking")
	private DisruptorWaitStrategy waitStrategy;
	@UriParam(label = "producer", defaultValue = "Multi")
	private DisruptorProducerType producerType;

	public DisruptorEndpoint(final String endpointUri, final Component component,
	final DisruptorReference disruptorReference, final int concurrentConsumers,
	final boolean multipleConsumers, boolean blockWhenFull) throws Exception {
	super(endpointUri, component);
	this.disruptorReference = disruptorReference;
	this.name = disruptorReference.getName();
	this.concurrentConsumers = concurrentConsumers;
	this.multipleConsumers = multipleConsumers;
	this.blockWhenFull = blockWhenFull;
	}

	@ManagedAttribute(description = "Queue name")
	public String getName() {
	return name;
	}

	@ManagedAttribute(description = "Buffer max capacity")
	public int getBufferSize() {
	return disruptorReference.getBufferSize();
	}

	@ManagedAttribute(description = "Remaining capacity in ring buffer")
	public long getRemainingCapacity() throws DisruptorNotStartedException {
	return getDisruptor().getRemainingCapacity();
	}

	@ManagedAttribute(description = "Amount of pending exchanges waiting for consumption in ring buffer")
	public long getPendingExchangeCount() throws DisruptorNotStartedException {
	return getDisruptor().getPendingExchangeCount();
	}

	/**
	* Number of concurrent threads processing exchanges.
	*/
	@ManagedAttribute(description = "Number of concurrent consumers")
	public int getConcurrentConsumers() {
	return concurrentConsumers;
	}

	@ManagedAttribute(description = "Option to specify whether the caller should wait for the async task to complete or not before continuing")
	public WaitForTaskToComplete getWaitForTaskToComplete() {
	return waitForTaskToComplete;
	}

	/**
	* Option to specify whether the caller should wait for the async task to complete or not before continuing.
	* The following three options are supported: Always, Never or IfReplyExpected. The first two values are self-explanatory.
	* The last value, IfReplyExpected, will only wait if the message is Request Reply based.
	*/
	public void setWaitForTaskToComplete(final WaitForTaskToComplete waitForTaskToComplete) {
	this.waitForTaskToComplete = waitForTaskToComplete;
	}

	@ManagedAttribute(description = "Timeout (in milliseconds) before a producer will stop waiting for an asynchronous task to complete")
	public long getTimeout() {
	return timeout;
	}

	/**
	* Timeout (in milliseconds) before a producer will stop waiting for an asynchronous task to complete.
	* You can disable timeout by using 0 or a negative value.
	*/
	public void setTimeout(final long timeout) {
	this.timeout = timeout;
	}

	@ManagedAttribute(description = "The maximum capacity of the Disruptors ringbuffer")
	public int getSize() {
	return size;
	}

	/**
	* The maximum capacity of the Disruptors ringbuffer
	* Will be effectively increased to the nearest power of two.
	* Notice: Mind if you use this option, then its the first endpoint being created with the queue name,
	* that determines the size. To make sure all endpoints use same size, then configure the size option
	* on all of them, or the first endpoint being created.
	*/
	public void setSize(int size) {
	this.size = size;
	}

	@Override
	@ManagedAttribute(description = "Specifies whether multiple consumers are allowed")
	public boolean isMultipleConsumersSupported() {
	return isMultipleConsumers();
	}

	/**
	* Specifies whether multiple consumers are allowed.
	* If enabled, you can use Disruptor for Publish-Subscribe messaging.
	* That is, you can send a message to the queue and have each consumer receive a copy of the message.
	* When enabled, this option should be specified on every consumer endpoint.
	*/
	public boolean isMultipleConsumers() {
	return multipleConsumers;
	}

	/**
	* Returns the current active consumers on this endpoint
	*/
	public Set<DisruptorConsumer> getConsumers() {
	return Collections.unmodifiableSet(consumers);
	}

	/**
	* Returns the current active producers on this endpoint
	*/
	public Set<DisruptorProducer> getProducers() {
	return Collections.unmodifiableSet(producers);
	}

	@ManagedAttribute
	public boolean isBlockWhenFull() {
	return blockWhenFull;
	}

	/**
	* Whether a thread that sends messages to a full Disruptor will block until the ringbuffer's capacity is no longer exhausted.
	* By default, the calling thread will block and wait until the message can be accepted.
	* By disabling this option, an exception will be thrown stating that the queue is full.
	*/
	public void setBlockWhenFull(boolean blockWhenFull) {
	this.blockWhenFull = blockWhenFull;
	}

	@ManagedAttribute(description = "Defines the strategy used by consumer threads to wait on new exchanges to be published")
	public DisruptorWaitStrategy getWaitStrategy() {
	return waitStrategy;
	}

	/**
	* Defines the strategy used by consumer threads to wait on new exchanges to be published.
	* The options allowed are:Blocking, Sleeping, BusySpin and Yielding.
	*/
	public void setWaitStrategy(DisruptorWaitStrategy waitStrategy) {
	this.waitStrategy = waitStrategy;
	}

	@ManagedAttribute(description = " Defines the producers allowed on the Disruptor")
	public DisruptorProducerType getProducerType() {
	return producerType;
	}

	/**
	* Defines the producers allowed on the Disruptor.
	* The options allowed are: Multi to allow multiple producers and Single to enable certain optimizations only
	* allowed when one concurrent producer (on one thread or otherwise synchronized) is active.
	*/
	public void setProducerType(DisruptorProducerType producerType) {
	this.producerType = producerType;
	}

	@Override
	public boolean isSingleton() {
	return true;
	}

	@Override
	public Producer createProducer() throws Exception {
	if (getProducers().size() == 1 && getDisruptor().getProducerType() == DisruptorProducerType.Single) {
	throw new IllegalStateException(
	"Endpoint can't support multiple producers when ProducerType SINGLE is configured");
	}
	return new DisruptorProducer(this, getWaitForTaskToComplete(), getTimeout(), isBlockWhenFull());
	}

	@Override
	public Consumer createConsumer(final Processor processor) throws Exception {
	return new DisruptorConsumer(this, processor);
	}

	@Override
	protected void doStart() throws Exception {
	// notify reference we are shutting down this endpoint
	disruptorReference.addEndpoint(this);
	super.doStart();
	}

	@Override
	protected void doStop() throws Exception {
	// notify reference we are shutting down this endpoint
	disruptorReference.removeEndpoint(this);
	super.doStop();
	}

	@Override
	protected void doShutdown() throws Exception {
	// notify component we are shutting down this endpoint
	if (getComponent() != null) {
	getComponent().onShutdownEndpoint(this);
	}

	super.doShutdown();
	}

	@Override
	public DisruptorComponent getComponent() {
	return (DisruptorComponent)super.getComponent();
	}

	void onStarted(final DisruptorConsumer consumer) throws Exception {
	synchronized (this) {
	// validate multiple consumers has been enabled is necessary
	if (!consumers.isEmpty() && !isMultipleConsumersSupported()) {
	throw new IllegalStateException(
	"Multiple consumers for the same endpoint is not allowed: " + this);
	}
	if (consumers.add(consumer)) {
	LOGGER.debug("Starting consumer {} on endpoint {}", consumer, getEndpointUri());
	getDisruptor().reconfigure();
	} else {
	LOGGER.debug("Tried to start Consumer {} on endpoint {} but it was already started", consumer, getEndpointUri());
	}
	}
	}


	void onStopped(final DisruptorConsumer consumer) throws Exception {
	synchronized (this) {
	if (consumers.remove(consumer)) {
	LOGGER.debug("Stopping consumer {} on endpoint {}", consumer, getEndpointUri());
	getDisruptor().reconfigure();
	} else {
	LOGGER.debug("Tried to stop Consumer {} on endpoint {} but it was already stopped", consumer, getEndpointUri());
	}
	}
	}

	void onStarted(final DisruptorProducer producer) {
	producers.add(producer);
	}

	void onStopped(final DisruptorProducer producer) {
	producers.remove(producer);
	}

	Map<DisruptorConsumer, Collection<LifecycleAwareExchangeEventHandler>> createConsumerEventHandlers() {
	Map<DisruptorConsumer, Collection<LifecycleAwareExchangeEventHandler>> result =
	new HashMap<DisruptorConsumer, Collection<LifecycleAwareExchangeEventHandler>>();

	for (final DisruptorConsumer consumer : consumers) {
	result.put(consumer, consumer.createEventHandlers(concurrentConsumers));
	}

	return result;
	}

	/**
	* Called by DisruptorProducers to publish new exchanges on the RingBuffer, blocking when full
	*/
	void publish(final Exchange exchange) throws DisruptorNotStartedException {
	disruptorReference.publish(exchange);
	}

	/**
	* Called by DisruptorProducers to publish new exchanges on the RingBuffer, throwing InsufficientCapacityException
	* when full
	*
	* @throws InsufficientCapacityException when the Ringbuffer is full.
	*/
	void tryPublish(final Exchange exchange) throws DisruptorNotStartedException, InsufficientCapacityException {
	disruptorReference.tryPublish(exchange);
	}

	DisruptorReference getDisruptor() {
	return disruptorReference;
	}

	@Override
	public boolean equals(Object object) {
	boolean result = super.equals(object);
	return result && getCamelContext().equals(((DisruptorEndpoint)object).getCamelContext());
	}

	@Override
	public int hashCode() {
	return getEndpointUri().hashCode() * 37 + getCamelContext().hashCode();
	}
	}