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

public Consumer createConsumer(Processor processor) throws Exception {
    ObjectHelper.notNull(feedUri, "feedUri");

    FeedPollingConsumer answer;
    if (isSplitEntries()) {
        answer = createEntryPollingConsumer(this, processor, filter, lastUpdate, throttleEntries);
    } else {
        answer = createPollingConsumer(this, processor);
    }

    // ScheduledPollConsumer default delay is 500 millis and that is too often for polling a feed,
    // so we override with a new default value. End user can override this value by providing a consumer.delay parameter
    answer.setDelay(FeedPollingConsumer.DEFAULT_CONSUMER_DELAY);
    configureConsumer(answer);
    return answer;
}

/**
 * Create one of the camel consumer available based on the configuration
 * @param processor  the given processor
 * @return the camel consumer
 * @throws Exception
 */
public Consumer createConsumer(Processor processor) throws Exception {
    LOG.trace("Resolve consumer dropbox endpoint {" + configuration.getOperation().toString() + "}");
    LOG.trace("Resolve consumer dropbox attached client:" + configuration.getClient());
    DropboxScheduledPollConsumer consumer = null;
    if (this.configuration.getOperation() == DropboxOperation.search) {
        consumer = new DropboxScheduledPollSearchConsumer(this, processor, configuration);
        consumer.setDelay(POLL_CONSUMER_DELAY);
        return consumer;
    } else if (this.configuration.getOperation() == DropboxOperation.get) {
        consumer = new DropboxScheduledPollGetConsumer(this, processor, configuration);
        consumer.setDelay(POLL_CONSUMER_DELAY);
        return consumer;
    } else {
        throw new DropboxException("Operation specified is not valid for consumer!");
    }
}

public void testCreateAbsoluteDirectory() throws Exception {
    deleteDirectory("target/file/foo");
    // use current dir as base as absolute path
    String base = new File("").getAbsolutePath() + "/target/file/foo";

    Endpoint endpoint = context.getEndpoint("file://" + base);
    Consumer consumer = endpoint.createConsumer(new Processor() {
        public void process(Exchange exchange) throws Exception {
            // noop
        }
    });

    consumer.start();
    consumer.stop();

    // the directory should now exists
    File dir = new File(base);
    assertTrue("Directory should be created", dir.exists());
    assertTrue("Directory should be a directory", dir.isDirectory());
}

public Consumer createConsumer(Processor processor) throws Exception {
    if (getComponent() != null) {
        // all consumers must match having the same multipleConsumers options
        String key = getComponent().getQueueKey(getEndpointUri());
        QueueReference ref = getComponent().getQueueReference(key);
        if (ref != null && ref.getMultipleConsumers() != isMultipleConsumers()) {
            // there is already a multiple consumers, so make sure they matches
            throw new IllegalArgumentException("Cannot use existing queue " + key + " as the existing queue multiple consumers "
                    + ref.getMultipleConsumers() + " does not match given multiple consumers " + multipleConsumers);
        }
    }

    Consumer answer = createNewConsumer(processor);
    configureConsumer(answer);
    return answer;
}

@Override
public Consumer createConsumer(Processor processor) throws Exception {
    TwitterConsumer twitter4jConsumer = TwitterHelper.createConsumer(this, getEndpointUri());
    TwitterConsumerDirect answer = new TwitterConsumerDirect(this, processor, twitter4jConsumer);
    configureConsumer(answer);
    return answer;
}

public Consumer createConsumer(Processor processor) throws Exception {
    switch (config.getFunctionType()) {
    case MESSAGES:
    case ALGO:
    case FOLLOWING:
    case MY_FEED:
    case PRIVATE:
    case SENT:
    case RECEIVED:
        return new YammerMessagePollingConsumer(this, processor);
    case USERS:
    case CURRENT:
        return new YammerUserPollingConsumer(this, processor);
    case RELATIONSHIPS:
        return new YammerRelationshipPollingConsumer(this, processor);
    default:
        throw new Exception(String.format("%s is not a valid Yammer function type.", config.getFunction()));
    }  

}

@Override
public Consumer createConsumer(CamelContext camelContext, Processor processor, String verb, String basePath, String uriTemplate,
                               String consumes, String produces, RestConfiguration configuration, Map<String, Object> parameters) throws Exception {
    // just use a seda endpoint for testing purpose
    String id;
    if (uriTemplate != null) {
        id = ActiveMQUuidGenerator.generateSanitizedId(basePath + uriTemplate);
    } else {
        id = ActiveMQUuidGenerator.generateSanitizedId(basePath);
    }
    // remove leading dash as we add that ourselves
    if (id.startsWith("-")) {
        id = id.substring(1);
    }
    SedaEndpoint seda = camelContext.getEndpoint("seda:" + verb + "-" + id, SedaEndpoint.class);
    return seda.createConsumer(processor);
}

public Consumer createConsumer(Processor processor) throws Exception {
    // make sure inBody is not set for consumers
    if (inBody != null) {
        throw new IllegalArgumentException("Option inBody is not supported for consumer endpoint");
    }

    // validate consumer APIs
    if (getApiName() != BoxApiName.POLL_EVENTS) {
        throw new IllegalArgumentException("Consumer endpoint only supports endpoint prefix "
            + BoxApiName.POLL_EVENTS.getName());
    }
    final BoxConsumer consumer = new BoxConsumer(this, processor);
    // also set consumer.* properties
    configureConsumer(consumer);
    return consumer;
}

public void testAutoCreateDirectoryWithDot() throws Exception {
    deleteDirectory("target/file/foo.bar");

    Endpoint endpoint = context.getEndpoint("file://target/file/foo.bar?autoCreate=true");
    Consumer consumer = endpoint.createConsumer(new Processor() {
        public void process(Exchange exchange) throws Exception {
            // noop
        }
    });

    consumer.start();
    consumer.stop();

    // the directory should exist
    File dir = new File("target/file/foo.bar");
    assertTrue("Directory should be created", dir.exists());
    assertTrue("Directory should be a directory", dir.isDirectory());
}

/**
 * Creates a consumer endpoint that splits up the List of Maps into exchanges of single
 * Maps, and within each exchange it converts each Map to JSON.
 */
@Override
public Consumer createConsumer(final Processor processor) throws Exception {
    final ToJSONProcessor toJsonProcessor = new ToJSONProcessor();
    Processor pipeline = Pipeline.newInstance(getCamelContext(), toJsonProcessor, processor);
    final Expression expression = ExpressionBuilder.bodyExpression(List.class);
    final Splitter splitter = new Splitter(getCamelContext(), expression, pipeline, null);
    return endpoint.createConsumer(splitter);
}

@Override
public Consumer createConsumer(final Processor processor) throws Exception {
    final Processor beforeConsumer = getBeforeConsumer();
    final Processor afterConsumer = getAfterConsumer();

    // use a pipeline to process before, processor, after in that order
    // create consumer with the pipeline
    final Processor pipeline = Pipeline.newInstance(getCamelContext(), beforeConsumer, processor, afterConsumer);
    final Consumer consumer = endpoint.createConsumer(pipeline);
    configureConsumer(consumer);

    return consumer;
}

@Override
public Consumer createConsumer(Processor processor) throws Exception {
    SpongeConsumer answer = new SpongeConsumer(this, processor, engine);
    configureConsumer(answer);
    return answer;
}

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

@Override
public Consumer createConsumer(Processor processor) throws Exception {

    return new OrientDBConsumer(this, processor);
}

public Consumer createConsumer(Processor processor) throws Exception {
    throw new UnsupportedOperationException("You cannot receive messages from this endpoint");
}

@Override
public Consumer createConsumer(Processor processor) throws Exception {
    HazelcastTopicConsumer answer = new HazelcastTopicConsumer(hazelcastInstance, this, processor, cacheName);
    configureConsumer(answer);
    return answer;
}

public void testCacheEndpoints() throws Exception {
    // test that we cache at most 1000 endpoints in camel context to avoid it eating to much memory
    for (int i = 0; i < 1234; i++) {
        String uri = "my:endpoint?id=" + i;
        DefaultEndpoint e = new DefaultEndpoint() {
            public Producer createProducer() throws Exception {
                return null;
            }
            public Consumer createConsumer(Processor processor) throws Exception {
                return null;
            }
            public boolean isSingleton() {
                return true;
            }
        };
        e.setCamelContext(context);
        e.setEndpointUri(uri);

        context.addEndpoint(uri, e);
    }

    // the eviction is async so force cleanup
    context.getEndpointRegistry().cleanUp();

    Collection<Endpoint> col = context.getEndpoints();
    assertEquals("Size should be 1000", 1000, col.size());
}

public Consumer createConsumer(Processor processor) throws Exception {
    HBaseConsumer consumer = new HBaseConsumer(this, processor);
    configureConsumer(consumer);
    consumer.setMaxMessagesPerPoll(maxMessagesPerPoll);
    return consumer;
}

public Consumer createConsumer(Processor processor) throws Exception {
    // make sure inBody is not set for consumers
    if (inBody != null) {
        throw new IllegalArgumentException("Option inBody is not supported for consumer endpoint");
    }
    // only read method is supported
    if (!READ_METHOD.equals(methodName) && !UREAD_METHOD.equals(methodName)) {
        throw new IllegalArgumentException("Only read method is supported for consumer endpoints");
    }
    final Olingo2Consumer consumer = new Olingo2Consumer(this, processor);
    // also set consumer.* properties
    configureConsumer(consumer);
    return consumer;
}

@Override
public Consumer createConsumer(Processor processor) {
    throw new UnsupportedOperationException("Consumer not supported for JcloudsComputeEndpoint!");
}

@Override
public Consumer createConsumer(Processor processor) throws Exception {
    return new DefaultConsumer(this, processor) {
        @Override
        protected void doStart() throws Exception {
            processors.add(getProcessor());
        }

        @Override
        protected void doStop() throws Exception {
            processors.remove(getProcessor());                   
        }
    };
}

@Override
public Consumer createConsumer(Processor processor) throws Exception {
    if (Jt400Type.DTAQ == configuration.getType()) {
        return new Jt400DataQueueConsumer(this);
    } else {
        throw new OperationNotSupportedException();
    }
}

public Consumer createConsumer(Processor processor) throws Exception {
    ObjectHelper.notNull(config, "config");
    ObjectHelper.notNull(cacheManagerFactory, "cacheManagerFactory");
    CacheConsumer answer = new CacheConsumer(this, processor, config);
    configureConsumer(answer);
    return answer;
}

public Consumer createConsumer(Processor processor) throws Exception {
    if (type == GitHubType.COMMIT) {
        ObjectHelper.notEmpty(branchName, "branchName", this);
        return new CommitConsumer(this, processor, branchName);
    } else if (type == GitHubType.PULLREQUEST) {
        return new PullRequestConsumer(this, processor);
    } else if (type == GitHubType.PULLREQUESTCOMMENT) {
        return new PullRequestCommentConsumer(this, processor);
    } else if (type == GitHubType.TAG) {
        return new TagConsumer(this, processor);
    }
    throw new IllegalArgumentException("Cannot create consumer with type " + type);
}

public Consumer createConsumer(Processor processor) throws Exception {
    // consumer requires a topicName, operation name must be the invalid topic name
    if (topicName == null) {
        throw new IllegalArgumentException(String.format("Invalid topic name %s, matches a producer operation name",
                operationName.value()));
    }

    final SalesforceConsumer consumer = new SalesforceConsumer(this, processor,
        getComponent().getSubscriptionHelper());
    configureConsumer(consumer);
    return consumer;
}

protected void configureConsumer(Consumer consumer) throws Exception {
    if (consumerProperties != null) {
        // use a defensive copy of the consumer properties as the methods below will remove the used properties
        // and in case we restart routes, we need access to the original consumer properties again
        Map<String, Object> copy = new HashMap<String, Object>(consumerProperties);

        // set reference properties first as they use # syntax that fools the regular properties setter
        EndpointHelper.setReferenceProperties(getCamelContext(), consumer, copy);
        EndpointHelper.setProperties(getCamelContext(), consumer, copy);

        // special consumer.bridgeErrorHandler option
        Object bridge = copy.remove("bridgeErrorHandler");
        if (bridge != null && "true".equals(bridge)) {
            if (consumer instanceof DefaultConsumer) {
                DefaultConsumer defaultConsumer = (DefaultConsumer) consumer;
                defaultConsumer.setExceptionHandler(new BridgeExceptionHandlerToErrorHandler(defaultConsumer));
            } else {
                throw new IllegalArgumentException("Option consumer.bridgeErrorHandler is only supported by endpoints,"
                        + " having their consumer extend DefaultConsumer. The consumer is a " + consumer.getClass().getName() + " class.");
            }
        }

        if (!this.isLenientProperties() && copy.size() > 0) {
            throw new ResolveEndpointFailedException(this.getEndpointUri(), "There are " + copy.size()
                + " parameters that couldn't be set on the endpoint consumer."
                + " Check the uri if the parameters are spelt correctly and that they are properties of the endpoint."
                + " Unknown consumer parameters=[" + copy + "]");
        }
    }
}

@Override
public Consumer createConsumer(Processor processor) throws Exception {
    synchronized (this) {
        if (consumer != null && processor != consumer.getProcessor()) {
            throw new Exception("PerformanceTestEndpoint doesn not support multiple consumers per Endpoint");
        }
        consumer = new PerformanceTestConsumer(this, processor);
    }
    return consumer;
}

@Override
public Consumer createConsumer(Processor processor) throws Exception {
    if (consumerFactory == null) {
        throw new IllegalArgumentException("No consumer for " + apiEndpoint);
    }

    return consumerFactory.create(this, configuration, processor);
}

private RestServiceEntry(Consumer consumer, String url, String baseUrl, String basePath, String uriTemplate, String method,
                         String consumes, String produces, String inType, String outType, String routeId, String description) {
    this.consumer = consumer;
    this.url = url;
    this.baseUrl = baseUrl;
    this.basePath = basePath;
    this.uriTemplate = uriTemplate;
    this.method = method;
    this.consumes = consumes;
    this.produces = produces;
    this.inType = inType;
    this.outType = outType;
    this.routeId = routeId;
    this.description = description;
}

/**
 * Creates a consumer using the given processor and sender
 */
public Consumer createConsumer(Processor processor, JavaMailSender sender) throws Exception {
    MailConsumer answer = new MailConsumer(this, processor, sender);
    answer.setHandleFailedMessage(configuration.isHandleFailedMessage());
    answer.setSkipFailedMessage(configuration.isSkipFailedMessage());
    answer.setMaxMessagesPerPoll(getMaxMessagesPerPoll());
    configureConsumer(answer);
    return answer;
}

public void subscribeMethod(Method method, Object bean, String beanName, String endpointUri, String endpointName, String endpointProperty) {
    // lets bind this method to a listener
    String injectionPointName = method.getName();
    Endpoint endpoint = getEndpointInjection(bean, endpointUri, endpointName, endpointProperty, injectionPointName, true);
    if (endpoint != null) {
        try {
            Processor processor = createConsumerProcessor(bean, method, endpoint);
            Consumer consumer = endpoint.createConsumer(processor);
            LOG.debug("Created processor: {} for consumer: {}", processor, consumer);
            startService(consumer, endpoint.getCamelContext(), bean, beanName);
        } catch (Exception e) {
            throw ObjectHelper.wrapRuntimeCamelException(e);
        }
    }
}

public Consumer createConsumer(Processor processor) throws Exception {
    // make sure inBody is not set for consumers
    if (inBody != null) {
        throw new IllegalArgumentException("Option inBody is not supported for consumer endpoint");
    }
    final GoogleDriveConsumer consumer = new GoogleDriveConsumer(this, processor);
    // also set consumer.* properties
    configureConsumer(consumer);
    return consumer;
}

@Override
public Consumer createApiConsumer(CamelContext camelContext, Processor processor,
                                  String contextPath, RestConfiguration configuration,
                                  Map<String, Object> parameters) throws Exception {
    // just use a seda endpoint for testing purpose
    String id = ActiveMQUuidGenerator.generateSanitizedId(contextPath);
    // remove leading dash as we add that ourselves
    if (id.startsWith("-")) {
        id = id.substring(1);
    }
    SedaEndpoint seda = camelContext.getEndpoint("seda:api:" + "-" + id, SedaEndpoint.class);
    return seda.createConsumer(processor);
}

public void testCacheEndpoints() throws Exception {
    // test that we cache at most 75 endpoints in camel context to avoid it eating to much memory
    for (int i = 0; i < 100; i++) {
        String uri = "my:endpoint?id=" + i;
        DefaultEndpoint e = new DefaultEndpoint() {
            public Producer createProducer() throws Exception {
                return null;
            }
            public Consumer createConsumer(Processor processor) throws Exception {
                return null;
            }
            public boolean isSingleton() {
                return true;
            }
        };
        e.setCamelContext(context);
        e.setEndpointUri(uri);

        context.addEndpoint(uri, e);
    }

    // the eviction is async so force cleanup
    context.getEndpointRegistry().cleanUp();

    Collection<Endpoint> col = context.getEndpoints();
    assertTrue("Size should be at most 75 was " + col.size(), col.size() <= 75);
}

@Override
public Consumer createConsumer(Processor processor) throws Exception {
    ObjectHelper.notNull(component, "component");
    WebsocketConsumer consumer = new WebsocketConsumer(this, processor);
    configureConsumer(consumer);
    return consumer;
}

@Override
public Consumer createConsumer(Processor processor) throws Exception {
    Consumer answer = new DataSetConsumer(this, processor);
    configureConsumer(answer);

    // expectedMessageCount((int) size);

    return answer;
}

public boolean begin(Consumer consumer, Endpoint endpoint) {
    event += "begin";
    if (counter++ == 0) {
        // deny polling at first call
        return false;
    }
    return true;
}

@Override
public Consumer createConsumer(Processor processor) throws Exception {
    TwitterConsumer twitter4jConsumer = TwitterHelper.createConsumer(this, getEndpointUri());
    // update the pulling lastID with sinceId
    twitter4jConsumer.setLastId(properties.getSinceId());
    TwitterConsumerPolling tc = new TwitterConsumerPolling(this, processor, twitter4jConsumer);
    configureConsumer(tc);
    return tc;
}

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

@Override
public Consumer createConsumer(Processor processor) throws Exception {
    throw new UnsupportedOperationException("The PLC4X endpoint doesn't support consumers.");
}