@Test public void throttleWithTimePeriod() throws Exception { ServiceDomain domain = new MockDomain(); domain.setProperty(CamelContextConfigurator.SHUTDOWN_TIMEOUT, "5"); initDomain(domain); QName name = new QName("testThrottleTimePeriod"); final ExchangeSink sink = new ExchangeSink(); final Service service = new MockService(name, new InOnlyService(), sink); final ServiceReference reference = new ServiceReferenceImpl(name, new InOnlyService(), null, null); Throttling throttle = new Throttling().setMaxRequests(1).setTimePeriod(60 * 1000); ServiceMetadataBuilder.update(reference.getServiceMetadata()).throttling(throttle); final ExchangeDispatcher dispatch = _provider.createDispatcher(reference); final int NUM_SENDS = 5; for (int i = 0; i < NUM_SENDS; i++) { new Thread(new Runnable() { public void run() { Exchange exchange = dispatch.createExchange(sink, ExchangePattern.IN_ONLY); exchange.consumer(reference, reference.getInterface().getOperation(ServiceInterface.DEFAULT_OPERATION)); exchange.provider(service, service.getInterface().getOperation(ServiceInterface.DEFAULT_OPERATION)); Message message = exchange.createMessage(); exchange.send(message); } }).start(); } Thread.sleep(4000); Assert.assertEquals("Received more than one message per minute - throttling policy violated!", 1, sink.getReceivedCount()); }
@Override public Exchange createExchange(ExchangeHandler replyHandler, ExchangePattern pattern) { DefaultExchange exchange = new DefaultExchange(_context, translate(pattern)); exchange.setIn(new CamelMessage(exchange)); return new CamelExchange(this, exchange, replyHandler); }
ExchangeDispatcher implements Dispatcher { @Override public Exchange createExchange(ExchangeHandler replyHandler, ExchangePattern pattern) { DefaultExchange exchange = new DefaultExchange(_context, translate(pattern)); exchange.setIn(new CamelMessage(exchange)); return new CamelExchange(this, exchange, replyHandler); } }
ExchangeDispatcher implements Dispatcher { @Override public Exchange createExchange(ExchangeHandler replyHandler, ExchangePattern pattern) { DefaultExchange exchange = new DefaultExchange(_context, translate(pattern)); exchange.setIn(new CamelMessage(exchange)); return new CamelExchange(this, exchange, replyHandler); } ExchangeDispatcher(ModelCamelContext context, ServiceReference reference); }
ExchangeDispatcher implements Dispatcher { @Override public Exchange createExchange(ExchangeHandler replyHandler, ExchangePattern pattern) { DefaultExchange exchange = new DefaultExchange(_context, translate(pattern)); exchange.setIn(new CamelMessage(exchange)); return new CamelExchange(this, exchange, replyHandler); } ExchangeDispatcher(ModelCamelContext context, ServiceReference reference); @Override ServiceReference getServiceReference(); @Override Exchange createExchange(ExchangeHandler replyHandler, ExchangePattern pattern); @Override void dispatch(final Exchange exchange); }
ExchangeDispatcher implements Dispatcher { @Override public Exchange createExchange(ExchangeHandler replyHandler, ExchangePattern pattern) { DefaultExchange exchange = new DefaultExchange(_context, translate(pattern)); exchange.setIn(new CamelMessage(exchange)); return new CamelExchange(this, exchange, replyHandler); } ExchangeDispatcher(ModelCamelContext context, ServiceReference reference); @Override ServiceReference getServiceReference(); @Override Exchange createExchange(ExchangeHandler replyHandler, ExchangePattern pattern); @Override void dispatch(final Exchange exchange);  }