Java 类org.apache.camel.Handler 实例源码

@Handler
public void execute(EventData eventData) {
  try {
    EventGenerator eg = EventGeneratorFactory.getEventGenerator(eventData.domainId,
        eventData.objectType);
    if(eventData.eventId == IEvent.DELETED) {
      DomainConfig dc = DomainConfig.getInstance(eventData.domainId);
      eg.generateDeleteEvent(eg, eventData.eventObject, eventData.objectId, dc.getLocale(), dc.getTimezone(), eventData.params);
    } else {
      eg.generate(eventData.eventId, eventData.params, eventData.objectId,
          eventData.customOptions);
    }
  } catch (Exception e) {
    log.severe("Issue with submitted task {0} with error {1}", eventData, e.getMessage(), e);
    throw new RuntimeException(e);
  }
}

@Handler
public String enhanceMessage( String body,  Exchange exchange  ) {

    String[] topicParts = exchange.getIn().getHeader("CamelMQTTSubscribeTopic", String.class).split(TOPIC_SEPARTOR);

    if(topicParts.length != 4) {
        throw new IllegalArgumentException("Invalid number of topic components. Expected " + TOPIC_PART_SIZE + ". Was " + topicParts.length);
    }

    StringBuilder sb = new StringBuilder();
    sb.append(body);
    sb.append(COMMA);
    sb.append(topicParts[1]);
    sb.append(COMMA);
    sb.append(topicParts[2]);
    sb.append(COMMA);
    sb.append(topicParts[3]);

    return sb.toString();
}

@Handler
public void processCsv(Exchange exchange) {
    // We assume Camel CSV is used beforehand to set the body to a List<List<String>>
    @SuppressWarnings("unchecked")
    List<List<String>> list = (List<List<String>>) exchange.getIn().getBody();

    List<String> tokens = list.get(0);

    // Build up a model for Velocity to use
    Map<String, Object> model = new HashMap<String, Object>();

    String[] nameParts = tokens.get(0).split(" ");
    model.put("firstName", nameParts[0]);
    model.put("lastName", nameParts[1]);
    model.put("birthDate", tokens.get(1));

    exchange.getIn().setBody(model);

    // Set the header required by ml-xcc for ingestion
    exchange.getIn().setHeader("MlXccUri", tokens.get(0) + ".xml");
}

@Handler
public FetchData fetch(ScrapeTemplate template,
                  Map<String, Object> actualParams) {
    LowerEnumSerializer.LOWER = false;
    final String uri = template.getUri();
    log.info("Fetching {} {} {} ...", template.getProtocol(), template.getProtocolVersion(),
            template.getUri());
    try {
        // FIXME: Use Retryer (with limit) when: Caused by: java.net.SocketException: Connection reset
        // at java.net.SocketInputStream.read(SocketInputStream.java:209)

        switch (template.getProtocol()) {
            case JSONRPC:
                return fetchJsonRpc(uri, template, actualParams);
            case HTTP:
                throw new UnsupportedOperationException("no HTTP support yet");
            default:
                throw new UnsupportedOperationException("Unsupported protocol: " + template.getProtocol());
        }
    } catch (Exception e) {
        throw new ScrapeException(e, "Cannot fetch %s", uri);
    }
}

/**
 * Return a {@link DsMetaData} response.
 * 
 * @param inputString
 *            input message body
 * @return meta data XML entity instance
 */
@Handler
@InOnly
public DsMetaData getDsMetaData(final String inputString)
{
    if (log.isDebugEnabled())
    {
        log.debug(getDataSourcePrefix() + "getDsMetaData()");
        // " inputString "
        // + inputString);
    }

    // For test assertion simplicity, the name field below does not depend on
    // the input string. Only the description field does.
    return new DsMetaData(name, name + ": " + inputString);
}

@Handler
public boolean lastFileProcessed(Exchange exchange) throws Exception {
    FileEntryWorkNote workNote = exchange.getIn().getBody(FileEntryWorkNote.class);

    String batchJobId = workNote.getBatchJobId();
    TenantContext.setJobId(batchJobId);

    if (batchJobDAO.updateFileEntryLatch(workNote.getBatchJobId(), workNote.getFileEntry().getFileName())) {

        RangedWorkNote wn = RangedWorkNote.createSimpleWorkNote(batchJobId);
        exchange.getIn().setBody(wn, RangedWorkNote.class);
        return true;

    }

    return false;
}

@Handler
public void processCommand(Exchange exch) throws Exception {
    String command = exch.getIn().getBody().toString();
    LOG.info("Received: " + command);
    String[] chunks = command.split("\\|");

    if (JOB_COMPLETED.equals(chunks[0])) {

        LOG.info("Clearing cache at job completion.");


        dumpAspectTrackers(chunks);

    } else {
        LOG.error("Unsupported command");
    }
}

@Handler
public List<String> getRetweetsWithScreenNames(@Header("statusId") Long statusId)
        throws TwitterException {
    return twitter.getRetweets(statusId).stream()
            .map(tweet -> tweet.getUser().getScreenName())
            .collect(Collectors.toList());
}

@Handler
public void log(@Body Object body){
    if(trace) {
        LOGGER.trace("Body is: {}",body);
    } else {
        LOGGER.info("Body is: {}",body);
    }
}

@Handler
public void execute(TaskOptions taskOptions) {
  jmsMeter.mark();
  log.info("Submitting task with parameter{0}", taskOptions);
  //localQueue.add(taskOptions);
  Task simpleTask = new SimpleTask(taskOptions);
  try {
    ((Runnable) simpleTask).run();
  } catch (Exception e) {
    log.severe("Issue with submitted task {0} with error {1}", taskOptions, e.getMessage(), e);
    throw e;
  }
}

@Handler
public void execute(AssetStatusRequest assetStatusRequest) throws ServiceException {

  jmsMeter.mark();

  List<AssetStatusModel> assetStatusModelList = assetStatusRequest.data;
  if (assetStatusModelList != null && !assetStatusModelList.isEmpty()) {
    AssetManagementService assetManagementService = Services
        .getService(AssetManagementServiceImpl.class);
    assetManagementService
        .updateAssetStatus(build(assetStatusModelList, assetManagementService));

  }

}

@Handler
public void process(@XPath("//title") String title,
                    @XPath("//description") String description,
                    @XPath("//pubDate") String pubDate,
                    @XPath("//link") String link) {
  ZonedDateTime published = ZonedDateTime.parse(pubDate, DateTimeFormatter.RFC_1123_DATE_TIME);
  if (ChronoUnit.MINUTES.between(published,
      ZonedDateTime.now()) <= threshold) {
    bus.broadcast(parser.parse(description, title, link))
        .as("status.redhat");
  }
}

@Handler
public void throwNonBpmnException() throws Exception {
    LOGGER.debug("throwing non bpmn bug");

    switch (getExceptionType()) {
    case NO_EXCEPTION:
        break;
    case NON_BPMN_EXCEPTION:
        throw new Exception("arbitrary non bpmn exception");
    case BPMN_EXCEPTION:
        throw new BpmnError("testError");
    }
}

@Handler
public void throwNonBpmnException() throws Exception {
    LOGGER.debug("throwing non bpmn bug");

    switch (getExceptionType()) {
    case NO_EXCEPTION:
        break;
    case NON_BPMN_EXCEPTION:
        throw new Exception("arbitrary non bpmn exception");
    case BPMN_EXCEPTION:
        throw new BpmnError("testError");
    }
}

@Handler
public void handler(@XPath("//response/text()") String response,
                    @XPath(headerName = "invoiceDetails", value = "//invoice/person/name/text()") String userName,
                    @XPath(headerName = "invoiceDetails", value = "//invoice/person/date", resultType = String.class) String date) {
    this.response = response;
    this.userName = userName;
    this.date = date;
}

@Handler
public String consume() {
    StringBuilder result = new StringBuilder();

    Exchange exchange;
    while ((exchange = consumer.receive("activemq:queue", 2000)) != null) {
        result.append(exchange.getIn().getBody(String.class));
    }

    return result.toString();

}

@Handler
public void handle(@Body CannotMarshal body) {
    if (body.getMessageNo() == 2) {
        // fail on second message
        throw new RuntimeException("Kaboom");
    }
}

@Handler
public void handler(@XQuery("/response") String response,
                    @XQuery(headerName = "invoiceDetails", value = "/invoice/person/name") String userName,
                    @XQuery(headerName = "invoiceDetails", value = "/invoice/person/date") String date) {
    this.response = response;
    this.userName = userName;
    this.date = date;
}

@Handler
public void orderSomeBooks() throws Exception {
    log.info("-------------------------------------------------------------------------------------------------------------------------");
    log.info("Make sure to have Postgres database up and running, as configured in the src/test/resources/META-INF/persistence.xml file");
    log.info("-------------------------------------------------------------------------------------------------------------------------");

    BookOrder order = new BookOrder();
    order.setAmount(1);
    order.setTitle("Camel in Action 2nd ed");

    template.sendBody("jpa:camelinaction.BookOrder", order);

    Thread.sleep(5000);
    log.info("... sleeping for 5 seconds and then stopping the route");

    // now stop the route
    context.stopRoute("books");

    // insert a new order which will sit in the database
    BookOrder order2 = new BookOrder();
    order2.setAmount(3);
    order2.setTitle("ActiveMQ in Action");

    template.sendBody("jpa:camelinaction.BookOrder", order2);

    log.info("-------------------------------------------------------------------------------------------------------------------------");
    log.info("Now we want to provoke a connection error, so stop the postgres database - and then press ENTER to continue!");
    log.info("-------------------------------------------------------------------------------------------------------------------------");

    System.console().readLine();

    context.startRoute("books");
    log.info("... starting route which should indicate some errors, which the bridge error handler should catch and handle");
    log.info("Notice that the consumer will backoff and not poll so fast, instead of every second, it now runs x10 sec.");
    log.info("Press CTRL+C to exit this application!");
}

@Handler
public void archive() {
  final List<Person> persons = personRepository.findAll();
  if ((persons == null) || persons.isEmpty()) {
    LOG.info("Nothing to backup");
    return;
  }

  LOG.info("Backup: {}", Archive.with(persons));
}

@Handler
public SyncHelloResponse composeGreeting(@Body SyncHelloRequest req) {
    Assert.notNull(req, "req must not be null");

    String greeting = "Hello " + req.getName();

    SyncHelloResponse res = new SyncHelloResponse();
    res.setGreeting(greeting);
    return res;
}

@Handler
public void printGreeting(@Body AsyncHelloRequest req) {
    Assert.notNull(req, "req must not be null");

    String greeting = "Hello " + req.getName();

    Log.debug("Greeting: " + greeting);
}

/**
 * Handles specified exception.
 *
 * @param ex the thrown exception
 * @param asynch {@code true} if it's asynchronous message processing otherwise synchronous processing
 * @return next route URI
 */
@Handler
public String exceptionHandling(Exception ex, @Header(AsynchConstants.ASYNCH_MSG_HEADER) Boolean asynch) {
    Assert.notNull(ex, "the ex must not be null");
    Assert.isTrue(BooleanUtils.isTrue(asynch), "it must be asynchronous message");

    String nextUri;

    if (ExceptionUtils.indexOfThrowable(ex, ValidationException.class) >= 0
            || ExceptionUtils.indexOfThrowable(ex, ValidationIntegrationException.class) >= 0) {
        Log.warn("Validation error, no further processing - " + ex.getMessage());
        nextUri = AsynchConstants.URI_ERROR_FATAL;

    } else if (ExceptionUtils.indexOfThrowable(ex, BusinessException.class) >= 0) {
        Log.warn("Business exception, no further processing.");
        nextUri = AsynchConstants.URI_ERROR_FATAL;

    } else if (ExceptionUtils.indexOfThrowable(ex, NoDataFoundException.class) >= 0) {
        Log.warn("No data found, no further processing.");
        nextUri = AsynchConstants.URI_ERROR_FATAL;

    } else if (ExceptionUtils.indexOfThrowable(ex, MultipleDataFoundException.class) >= 0) {
        Log.warn("Multiple data found, no further processing.");
        nextUri = AsynchConstants.URI_ERROR_FATAL;

    } else if (ExceptionUtils.indexOfThrowable(ex, LockFailureException.class) >= 0) {
        Log.warn("Locking exception.");
        nextUri = AsynchConstants.URI_ERROR_HANDLING;

    } else {
        Log.error("Unspecified exception - " + ex.getClass().getSimpleName() + " (" + ex.getMessage() + ")");
        nextUri = AsynchConstants.URI_ERROR_HANDLING;
    }

    return nextUri;
}

/**
 * Checks if ESB goes down or not.
 *
 * @return {@code true} if ESB is in "stopping mode" otherwise {@code false}
 */
@Handler
public boolean isNotInStoppingMode() {
    StopService stopService = getApplicationContext().getBean(StopService.class);

    Log.debug("ESB stopping mode is switched on: " + stopService.isStopping());

    return !stopService.isStopping();
}

/**
 * Checks if ESB goes down or not.
 *
 * @return {@code true} if ESB is in "stopping mode" otherwise {@code false}
 */
@Handler
public boolean isNotInStoppingMode() {
    StopService stopService = getApplicationContext().getBean(StopService.class);

    Log.debug("ESB stopping mode is switched on: " + stopService.isStopping());

    return !stopService.isStopping();
}

/**
 * Checks if current message wasn't converted to other state or is being processed more times.
 * It can happen when message is long time in queue that repairing process converts message back
 * to PARTLY_FAILED state and evenly message can start with duplicate processing.
 *
 * @param msg the message
 * @return {@code true} when message is obsolete otherwise {@code false}
 */
@Handler
public boolean isMessageObsolete(@Body Message msg) {
    Assert.notNull(msg, "the msg must not be null");
    MessageService messageService = getBean(MessageService.class);

    Message dbMsg = messageService.findMessageById(msg.getMsgId());

    Assert.notNull(dbMsg, "there must be message with ID=" + msg.getMsgId());

    return dbMsg.getState() != MsgStateEnum.PROCESSING
            && dbMsg.getLastUpdateTimestamp().before(msg.getLastUpdateTimestamp());
}

@Handler
public void logStartProcessing(@Body Message msg,
        @Nullable @Header(AsynchConstants.MSG_QUEUE_INSERT_HEADER) Long msgInsertTime) {

    Log.debug("Starts processing of the message {}, waited in queue for {} ms", msg.toHumanString(),
            msgInsertTime != null ? (System.currentTimeMillis() - msgInsertTime) : "-");
}

/**
 * Sets {@link AsynchConstants#OBJECT_ID_HEADER} and {@link AsynchConstants#ENTITY_TYPE_HEADER}
 * headers if there are available corresponding values in message.
 *
 * @param msg the message
 * @param headers the headers
 */
@Handler
public void setEntityInfo(@Body Message msg, @Headers Map<String, Object> headers) {
    Assert.notNull(msg, "the msg must not be null");

    if (msg.getObjectId() != null) {
        headers.put(OBJECT_ID_HEADER, msg.getObjectId());
    }
    if (msg.getEntityType() != null) {
        headers.put(ENTITY_TYPE_HEADER, msg.getEntityType());
    }
}

/**
 * Returns {@code true} if failed count exceeds limit for failing.
 *
 * @param msg the message
 * @return {@code true} when limit was exceeded, otherwise {@code false}
 */
@Handler
public boolean checkFailedLimit(@Header(MSG_HEADER) Message msg) {
    Assert.notNull(msg, "the msg must not be null");

    return msg.getFailedCount() >= countPartlyFailsBeforeFailed;
}

/**
  * Sends notification mail to admin(s).
  */
 @Handler
 public void sendMailToAdmin(@Header(MSG_HEADER) Message msg) {
     Assert.notNull(msg, "the msg must not be null");

     EmailService emailService = lookup(EmailService.BEAN, EmailService.class);

     String body = "The following message " + msg.toHumanString() + " FAILED.\n\nBody:\n" + msg.getEnvelope();

     emailService.sendEmailToAdmins("Notification about FAILED message", body);
}

/**
 * Makes call.
 *
 * @param callId Call ID for getting call parameters from {@link ContextCallRegistry}
 */
@Handler
public void makeCall(@Header(CALL_ID_HEADER) String callId) {
    Assert.hasText(callId, "the callId must not be empty");

    // get params
    ContextCallParams params = callRegistry.getParams(callId);

    Object res = ReflectionCallUtils.invokeMethod(params, getApplicationContext());

    // save response
    callRegistry.addResponse(callId, res);

    Log.debug("Response of the call ID '" + callId + "' was saved: " + res);
}

/**
 * Sets request header if available.
 *
 * @param callId Call ID for getting call parameters from {@link DirectCallRegistry}
 * @param exchange Camel exchange
 */
@Handler
public void setHeader(@Header(CALL_ID_HEADER) String callId, Exchange exchange) {
    Assert.hasText(callId, "the callId must not be empty");

    DirectCallParams params = callRegistry.getParams(callId);

    if (params.getHeader() != null) {
        Log.debug("Direct WS call: header=" + params.getHeader());

        exchange.getIn().setHeader(SpringWebserviceConstants.SPRING_WS_SOAP_HEADER, params.getHeader());
    }
}

/**
 * Gets request to external system.
 *
 * @param callId Call ID for getting call parameters from {@link DirectCallRegistry}
 * @return request
 */
@Handler
public Object getRequest(@Header(CALL_ID_HEADER) String callId) {
    Assert.hasText(callId, "the callId must not be empty");

    DirectCallParams params = callRegistry.getParams(callId);

    Log.debug("Direct WS call: uri= " + params.getUri() + ",\nsenderRef= " + params.getSenderRef()
            + ",\nsoapAction= " + params.getSoapAction() + ",\nbody: " + params.getBody());

    return params.getBody();
}

/**
 * Gets URI for calling external system.
 *
 * @param callId Call ID for getting call parameters from {@link DirectCallRegistry}
 * @return WS URI
 */
@Handler
public String getWsUri(@Header(CALL_ID_HEADER) String callId) {
    Assert.hasText(callId, "the callId must not be empty");

    DirectCallParams params = callRegistry.getParams(callId);

    return getOutWsUri(params.getUri(), params.getSenderRef(), params.getSoapAction());
}

@Handler
public Map getProcessVariables(@Body String body,
                               @Header(Exchange.FILE_NAME) String filename,
                               @Simple("${date:now:yyyy-MM-dd kk:mm:ss}") String timestamp) {
    Map<String, Object> variables = new HashMap<String, Object>();
    variables.put("message", body);
    variables.put("orderid", filename);
    variables.put("timestamp", timestamp);
    return variables;
}

/**
 * Process a string body.
 * 
 * @param inputString
 *            input message body
 * @return processed input string. Contains this data source's details for easy test
 *         assertions
 */
@Handler
@InOnly
public String process(final String inputString)
{
    if (log.isTraceEnabled())
    {
        log.trace(getDataSourcePrefix() + "process() " + inputString);
    }
    return name + ": " + inputString;
}

/**
 * @param inputString
 * @return processed input string
 */
@Handler
public QueryContext processQueryContext(final @Body QueryContext parentContext)
{
    Validate.notNull(parentContext);
    if (log.isDebugEnabled())
    {
        log.debug(getDataSourcePrefix() + "processQueryContext() " + parentContext);
        log.debug("QC Data Source ID: " + parentContext.getDataSourceId());
        log.debug("Search criteria: " + parentContext.getQuery());
    }

    // This section mocks a data source's creation of a new child context, queuing it
    // and setting its parent.
    QueryContextEntity childContext = QueryContextEntity
            .newInstanceWithExecutionId();
    childContext.setDataSourceId(name);
    childContext.setParent(QueryContextEntity.newCopy(parentContext));
    // Assuming the search query's first criterion is of type [EQ,
    // desiredNumRecordsBackFromThisDataSource, value]. Set value (parameter #2
    // (0-based index) on the parameter list)as our output records.
    //
    // If there are two criteria, using the second criterion's value instead.
    final Long numRecords = getNumRecordsFromQuery(parentContext);

    childContext.setNumRecords(numRecords.longValue());
    childContext = (QueryContextEntity) contextService.queue(childContext);
    childContext.start();
    childContext.finish();

    final long sleepValue = getSleepValue();
    if (log.isDebugEnabled())
    {
        log.debug("Sleeping for " + sleepValue + " miliseconds");
    }
    IoUtil.sleep(sleepValue);

    return QueryContextToImpl.newCopy(childContext);
}

/**
 * @param commandTypeLabel
 * @param input
 * @return
 * @see edu.utah.further.ds.api.lifecycle.controller.LifeCycleController#triggerCommand(java.lang.String,
 *      java.io.Serializable)
 */
@Override
// Optional here, yet a useful annotation
@Handler
public Object triggerCommand(
        @Header(MessageHeader.COMMAND_TYPE_NAME) final String commandTypeLabel,
        @Body final Serializable input)
{
    return triggerCommandHelper(commandTypeLabel, input);
}

@Override
@Handler
public void onReceive(Object message)
{
    if (!(message instanceof Heartbeat))
    {
       logger.warn("Heartbeat message expected but received: " + message.toString());
       return;
    }
    heartbeatDao.record((Heartbeat) message);
}

@Handler
public String configure(@Body String body, @Header("ExtensionHeader") String header) {
    return String.join("-", body, header, message);
}