Java 类org.apache.camel.management.event.AbstractExchangeEvent 实例源码

@Override
public void notify(EventObject event) throws Exception {
    boolean covered = false;

    if (event instanceof AbstractExchangeEvent) {
        AbstractExchangeEvent ev = AbstractExchangeEvent.class.cast(event);

        final Exchange exchange = ev.getExchange();
        String metricPrefix = exchange.getFromRouteId();

        // if we can't find the prefix for the metrics then don't capture any
        if (metricPrefix == null || metricPrefix.equals("")) { return; }

        if (ev instanceof ExchangeCompletedEvent || ev instanceof ExchangeFailedEvent || ev instanceof ExchangeRedeliveryEvent) {
            onExchangeCompletedEvent(ev, metricPrefix);
            covered = true;
        } else {
            metrics.meter(name(event.getClass(), metricPrefix)).mark();
        }
    }

    if (!covered)
        LOGGER.debug("Not covered: Type {} ({})", event.getClass(), event);
}

@Override
protected void setUp() throws Exception {
    super.setUp();

    breakpoint = new BreakpointSupport() {
        public void onEvent(Exchange exchange, EventObject event, ProcessorDefinition<?> definition) {
            AbstractExchangeEvent aee = (AbstractExchangeEvent) event;
            Exception e = aee.getExchange().getException();
            logs.add("Breakpoint at " + definition + " caused by: " + e.getClass().getSimpleName() + "[" + e.getMessage() + "]");
        }
    };

    exceptionCondition = new ConditionSupport() {
        public boolean matchEvent(Exchange exchange, EventObject event) {
            return event instanceof ExchangeFailedEvent;
        }
    };
}

private <T extends EventObject> void camelEventNotifiers(@Observes ProcessObserverMethod<T, ?> pom) {
    // Only activate Camel event notifiers for explicit Camel event observers, that is, an observer method for a super type won't activate notifiers.
    Type type = pom.getObserverMethod().getObservedType();
    // Camel events are raw types
    if (type instanceof Class && Class.class.cast(type).getPackage().equals(AbstractExchangeEvent.class.getPackage())) {
        Set<Annotation> qualifiers = pom.getObserverMethod().getObservedQualifiers();
        if (qualifiers.isEmpty()) {
            eventQualifiers.add(ANY);
        } else if (qualifiers.size() == 1 && qualifiers.stream()
            .filter(isAnnotationType(Named.class)).findAny().isPresent()) {
            eventQualifiers.add(DEFAULT);
        } else {
            eventQualifiers.addAll(qualifiers);
        }
    }
}

void notify(T event) {
    logger.debug("Consuming CDI event [{}] with {}", event, this);

    Exchange exchange = getEndpoint().createExchange();
    // TODO: would that be possible to propagate the event metadata?
    exchange.getIn().setBody(event);

    // Avoid infinite loop of exchange events
    if (event instanceof AbstractExchangeEvent) {
        exchange.setProperty(Exchange.NOTIFY_EVENT, Boolean.TRUE);
    }
    try {
        getProcessor().process(exchange);
    } catch (Exception cause) {
        throw new RuntimeExchangeException("Error while processing CDI event", exchange, cause);
    } finally {
        if (event instanceof AbstractExchangeEvent) {
            exchange.setProperty(Exchange.NOTIFY_EVENT, Boolean.FALSE);
        }
    }
}

void notify(T event) {
    logger.debug("Consuming CDI event [{}] with {}", event, this);

    Exchange exchange = getEndpoint().createExchange();
    // TODO: propagate the event metadata
    exchange.getIn().setBody(event);

    // Avoid infinite loop of exchange events
    if (event instanceof AbstractExchangeEvent)
        exchange.setProperty(Exchange.NOTIFY_EVENT, Boolean.TRUE);

    try {
        getProcessor().process(exchange);
    } catch (Exception cause) {
        throw new RuntimeExchangeException("Error while processing CDI event", exchange, cause);
    } finally {
        if (event instanceof AbstractExchangeEvent)
            exchange.setProperty(Exchange.NOTIFY_EVENT, Boolean.FALSE);
    }
}

@Override
public void notify(EventObject event) throws Exception {
    AbstractExchangeEvent aee = (AbstractExchangeEvent) event;
    Exchange exchange = aee.getExchange();
    onEvent(exchange, event);

    if (event instanceof ExchangeCompletedEvent) {
        // fail safe to ensure we remove single steps when the Exchange is complete
        singleSteps.remove(exchange.getExchangeId());
    }
}

private <T extends EventObject> void camelEventNotifiers(@Observes ProcessObserverMethod<T, ?> pom) {
    // Only activate Camel event notifiers for explicit Camel event observers, that is,
    // an observer method for a super type won't activate notifiers.
    Type type = pom.getObserverMethod().getObservedType();
    // Camel events are raw types
    if (type instanceof Class && Class.class.cast(type).getPackage().equals(AbstractExchangeEvent.class.getPackage())) {
        Set<Annotation> qualifiers = pom.getObserverMethod().getObservedQualifiers();
        if (qualifiers.isEmpty())
            eventQualifiers.add(ANY);
        else if (qualifiers.size() == 1 && qualifiers.stream().anyMatch(isAnnotationType(Named.class)))
            eventQualifiers.add(DEFAULT);
        else
            eventQualifiers.addAll(qualifiers);
    }
}

protected void onExchangeCompletedEvent(AbstractExchangeEvent event, String metricPrefix) {
    Period p = new Period(event.getExchange().getProperty(Exchange.CREATED_TIMESTAMP, Date.class).getTime(), System.currentTimeMillis());

    metrics.timer(name(event.getClass(), metricPrefix)).update(p.getMillis(), TimeUnit.MILLISECONDS);
}

@Override
public boolean isEnabled(EventObject event) {
    return event instanceof AbstractExchangeEvent;
}

private void onAnyExchangeEvent(@Observes AbstractExchangeEvent event, @Named("anyContext") List<Class> events) {
    events.add(event.getClass());
}

private void onDefaultExchangeEvent(@Observes @Default AbstractExchangeEvent event, List<Class> events) {
    events.add(event.getClass());
}

private void onFirstExchangeEvent(@Observes @ContextName("first") AbstractExchangeEvent event, @ContextName("first") List<Class> events) {
    events.add(event.getClass());
}

private void onSecondExchangeEvent(@Observes @ContextName("second") AbstractExchangeEvent event, @ContextName("second") List<Class> events) {
    events.add(event.getClass());
}

private void onExchangeEvent(@Observes AbstractExchangeEvent event, List<Class> events) {
    events.add(event.getClass());
}

private void onAnyExchangeEvent(@Observes AbstractExchangeEvent event, @Named("anyContext") List<Class> events) {
    events.add(event.getClass());
}

private void onDefaultExchangeEvent(@Observes @Default AbstractExchangeEvent event, @Named("defaultContext") List<Class> events) {
    events.add(event.getClass());
}

private void onFirstExchangeEvent(@Observes @ContextName("first") AbstractExchangeEvent event, @ContextName("first") List<Class> events) {
    events.add(event.getClass());
}

private void onSecondExchangeEvent(@Observes @ContextName("second") AbstractExchangeEvent event, @ContextName("second") List<Class> events) {
    events.add(event.getClass());
}

private void onExchangeEvent(@Observes AbstractExchangeEvent event, List<Class> events) {
    events.add(event.getClass());
}