001 /**
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements. See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License. You may obtain a copy of the License at
008 *
009 * http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017 package org.apache.camel.component.dataset;
018
019 import java.util.concurrent.ExecutorService;
020
021 import org.apache.camel.CamelContext;
022 import org.apache.camel.Exchange;
023 import org.apache.camel.Processor;
024 import org.apache.camel.impl.DefaultConsumer;
025 import org.apache.camel.processor.ThroughputLogger;
026 import org.apache.camel.util.CamelLogger;
027 import org.apache.camel.util.URISupport;
028
029 /**
030 * DataSet consumer.
031 *
032 * @version
033 */
034 public class DataSetConsumer extends DefaultConsumer {
035 private final CamelContext camelContext;
036 private DataSetEndpoint endpoint;
037 private Processor reporter;
038 private ExecutorService executorService;
039
040 public DataSetConsumer(DataSetEndpoint endpoint, Processor processor) {
041 super(endpoint, processor);
042 this.endpoint = endpoint;
043 this.camelContext = endpoint.getCamelContext();
044 }
045
046 @Override
047 protected void doStart() throws Exception {
048 super.doStart();
049
050 if (reporter == null) {
051 reporter = createReporter();
052 }
053 final DataSet dataSet = endpoint.getDataSet();
054 final long preloadSize = endpoint.getPreloadSize();
055
056 sendMessages(0, preloadSize);
057 executorService = camelContext.getExecutorServiceManager().newSingleThreadExecutor(this, endpoint.getEndpointUri());
058
059 executorService.execute(new Runnable() {
060 public void run() {
061 if (endpoint.getInitialDelay() > 0) {
062 try {
063 Thread.sleep(endpoint.getInitialDelay());
064 } catch (InterruptedException e) {
065 Thread.currentThread().interrupt();
066 return;
067 }
068 }
069
070 sendMessages(preloadSize, dataSet.getSize());
071 }
072 });
073 }
074
075 @Override
076 protected void doStop() throws Exception {
077 super.doStop();
078
079 if (executorService != null) {
080 camelContext.getExecutorServiceManager().shutdown(executorService);
081 executorService = null;
082 }
083 }
084
085 protected void sendMessages(long startIndex, long endIndex) {
086 try {
087 for (long i = startIndex; i < endIndex; i++) {
088 Exchange exchange = endpoint.createExchange(i);
089 getProcessor().process(exchange);
090
091 try {
092 long delay = endpoint.getProduceDelay();
093 if (delay > 0) {
094 Thread.sleep(delay);
095 }
096 } catch (InterruptedException e) {
097 Thread.currentThread().interrupt();
098 break;
099 }
100 if (reporter != null) {
101 reporter.process(exchange);
102 }
103 }
104 } catch (Exception e) {
105 handleException(e);
106 }
107 }
108
109 protected ThroughputLogger createReporter() {
110 // must sanitize uri to avoid logging sensitive information
111 String uri = URISupport.sanitizeUri(endpoint.getEndpointUri());
112 CamelLogger logger = new CamelLogger(uri);
113 ThroughputLogger answer = new ThroughputLogger(logger, (int) endpoint.getDataSet().getReportCount());
114 answer.setAction("Sent");
115 return answer;
116 }
117 }