Java 类org.apache.camel.component.direct.DirectEndpoint 实例源码

private void sendExchangesThroughDroppingThrottler(List<Exchange> sentExchanges, int messages) throws Exception {
    ProducerTemplate myTemplate = context.createProducerTemplate();

    DirectEndpoint targetEndpoint = resolveMandatoryEndpoint("direct:sample", DirectEndpoint.class);
    for (int i = 0; i < messages; i++) {
        Exchange e = targetEndpoint.createExchange();
        e.getIn().setBody("<message>" + i + "</message>");
        // only send if we are still started
        if (context.getStatus().isStarted()) {
            myTemplate.send(targetEndpoint, e);
            sentExchanges.add(e);
            Thread.sleep(100);
        }
    }
    myTemplate.stop();
}

@Test
public void read() throws Exception {
    final ProducerTemplate tpl = CamelBridge.CONTEXT.createProducerTemplate();

    final JobOperator jobOperator = BatchRuntime.getJobOperator();

    final long id = jobOperator.start("camel-reader", new Properties());

    while (DirectEndpoint.class.cast(CamelBridge.CONTEXT.getEndpoint("direct:reader")).getConsumer() == null) {
        Thread.sleep(100);
    }

    tpl.sendBody("direct:reader", "input#1");
    tpl.sendBody("direct:reader", null);

    Batches.waitForEnd(jobOperator, id);

    assertEquals(StoreItems.ITEMS.size(), 1);
    assertEquals("input#1", StoreItems.ITEMS.get(0));
}

private NotifyBuilder fromRoutesOnly() {
    // internal and should always be in top of stack
    stack.add(0, new EventPredicateSupport() {

        @Override
        public boolean isAbstract() {
            // is abstract as its a filter
            return true;
        }

        @Override
        public boolean onExchange(Exchange exchange) {
            // always accept direct endpoints as they are a special case as it will create the UoW beforehand
            // and just continue to route that on the consumer side, which causes the EventNotifier not to
            // emit events when the consumer received the exchange, as its already done. For example by
            // ProducerTemplate which creates the UoW before producing messages.
            if (exchange.getFromEndpoint() != null && exchange.getFromEndpoint() instanceof DirectEndpoint) {
                return true;
            }
            return EndpointHelper.matchPattern(exchange.getFromRouteId(), "*");
        }

        public boolean matches() {
            // should be true as we use the onExchange to filter
            return true;
        }

        @Override
        public String toString() {
            // we dont want any to string output as this is an internal predicate to match only from routes
            return "";
        }
    });
    return this;
}

public void testManagedEndpoint() throws Exception {
    // JMX tests dont work well on AIX CI servers (hangs them)
    if (isPlatform("aix")) {
        return;
    }

    assertDefaultDomain();

    resolveMandatoryEndpoint("direct:start", DirectEndpoint.class);

    ObjectName objName = new ObjectName(domainName + ":type=endpoints,*");
    Set<ObjectName> s = mbsc.queryNames(objName, null);
    assertEquals(2, s.size());
}

@Override
public void dispatch(final Exchange exchange) {
    // We can only send Camel exchanges through the camel bus
    if (!CamelExchange.class.isInstance(exchange)) {
        throw BusMessages.MESSAGES.onlyCamelExchanges();
    }
    CamelExchange camelEx = (CamelExchange)exchange;

    // For camel exchanges, the only phase we care about is IN.  The dispatch method can also
    // be called on the OUT path, but that should be handled by the IN_OUT filter in the Camel
    // bus route.
    if (!exchange.getPhase().equals(ExchangePhase.IN)) {
        return;
    }

    Throttling throttling = _reference.getServiceMetadata().getThrottling();
    if (throttling != null && throttling.getMaxRequests() > 0) {
        exchange.getMessage().getContext().setProperty(
                Throttling.MAX_REQUESTS, throttling.getMaxRequests())
                .addLabels(BehaviorLabel.TRANSIENT.label());
    }

    String uri = "direct:" + exchange.getConsumer().getName();
    DirectEndpoint ep = _context.getEndpoint(uri, DirectEndpoint.class);
    if (ep.getConsumer() == null) {
        throw BusMessages.MESSAGES.consumerForinternalCamelRouteNotFound(uri);
    }
    _producer.send(ep, camelEx.getExchange());

}

@Test
public void write() throws Exception {
    final ConsumerTemplate tpl = CamelBridge.CONTEXT.createConsumerTemplate();
    final Collection<Object> received = new ArrayList<Object>(2);

    final ExecutorService thread = Executors.newSingleThreadExecutor();
    thread.submit(new Runnable() {
        @Override
        public void run() {
            do {
                received.add(tpl.receiveBody("direct:writer"));
            } while (received.size() < 2);
        }
    });
    thread.shutdown();

    do { // starting the listener in another thread w can get timing issues so ensuring we are in the right state
        Thread.sleep(20);
    } while (DirectEndpoint.class.cast(CamelBridge.CONTEXT.getEndpoint("direct:writer")).getConsumer() == null);

    final JobOperator jobOperator = BatchRuntime.getJobOperator();
    Batches.waitForEnd(jobOperator, jobOperator.start("camel-writer", new Properties()));

    thread.awaitTermination(5, TimeUnit.MINUTES); // wait end of the thread before checking received

    assertTrue(received.contains("1"), received.toString());
    assertTrue(received.contains("2"), received.toString());
}

@Test
public void queryGets1ValueBack() throws Exception {
    setupManagementStrategy();

    Set<ObjectName> objectNames = new HashSet<ObjectName>();
    objectNames.add(new ObjectName("org.apache.camel:type=routes,name=route1,*"));
    objectNames.add(new ObjectName("org.apache.camel:type=routes,name=route2,*"));

    MBeanServer mBeanServerMocked = Mockito.mock(MBeanServer.class);
    provideRouteStats(mBeanServerMocked, objectNames);
    provideAllRoutes(mBeanServerMocked, objectNames);

    //TODO: update as failing test

    DirectComponent component = new DirectComponent();
    component.setCamelContext(context);

    DefaultChannel processor1 = new DefaultChannel();
    processor1.setChildDefinition(new ToDefinition(new DirectEndpoint("direct:lb1", component)));

    DefaultChannel processor2 = new DefaultChannel();
    processor2.setChildDefinition(new ToDefinition(new DirectEndpoint("direct:lb2", component)));

    List<Processor> processors = new LinkedList<Processor>();
    processors.add(processor1);
    processors.add(processor2);

    MBeanRouteStatisticsCollector collector = new MBeanRouteStatisticsCollector(context, mBeanServerMocked, "route", false, false);
    List<RouteStatistics> stats = collector.query(processors, createExchange());

    Assert.assertNotNull(stats);
    Assert.assertEquals(2, stats.size());

    RouteStatistics first = stats.get(0);
    Assert.assertEquals(new Long(1), new Long(first.getInflightExchange()));
    Assert.assertEquals(new Long(20L), new Long(first.getMeanProcessingTime()));
    Assert.assertEquals(new Long(30L), new Long(first.getLastProcessingTime()));
    Assert.assertEquals("1", first.getLoad01());
    Assert.assertEquals("5", first.getLoad05());
    Assert.assertEquals("15", first.getLoad15());
}

public V1CamelServiceBindingModelTest() {
    super(DirectEndpoint.class, CAMEL_XML);
}

public V1CamelDirectBindingModelTest() {
    super(DirectEndpoint.class, CAMEL_XML);
}

public V1CamelReferenceBindingModelTest() {
    super(DirectEndpoint.class, CAMEL_XML);
}