You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ac...@apache.org on 2016/09/28 08:15:36 UTC
[2/2] camel git commit: CAMEL-10331, Restore Docker Consumers
CAMEL-10331, Restore Docker Consumers
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/e27aaedd
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/e27aaedd
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/e27aaedd
Branch: refs/heads/master
Commit: e27aaedd8caeebf615b8c2c048a8edd74085c7b7
Parents: d1a512c
Author: Fabryprog <fa...@gmail.com>
Authored: Sun Sep 25 16:14:57 2016 +0200
Committer: Andrea Cosentino <an...@gmail.com>
Committed: Wed Sep 28 10:12:43 2016 +0200
----------------------------------------------------------------------
.../camel/component/docker/DockerEndpoint.java | 14 ++-
.../docker/consumer/DockerEventsConsumer.java | 109 +++++++++++++++++++
.../docker/consumer/DockerStatsConsumer.java | 99 +++++++++++++++++
.../consumer/DockerEventsConsumerTest.java | 89 +++++++++++++++
.../consumer/DockerStatsConsumerTest.java | 87 +++++++++++++++
.../component/docker/util/DockerTestUtils.java | 62 +++++++++++
6 files changed, 458 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/e27aaedd/components/camel-docker/src/main/java/org/apache/camel/component/docker/DockerEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-docker/src/main/java/org/apache/camel/component/docker/DockerEndpoint.java b/components/camel-docker/src/main/java/org/apache/camel/component/docker/DockerEndpoint.java
index 22f3ca0..75d54d4 100644
--- a/components/camel-docker/src/main/java/org/apache/camel/component/docker/DockerEndpoint.java
+++ b/components/camel-docker/src/main/java/org/apache/camel/component/docker/DockerEndpoint.java
@@ -16,10 +16,11 @@
*/
package org.apache.camel.component.docker;
-import org.apache.camel.CamelException;
import org.apache.camel.Consumer;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
+import org.apache.camel.component.docker.consumer.DockerEventsConsumer;
+import org.apache.camel.component.docker.consumer.DockerStatsConsumer;
import org.apache.camel.component.docker.exception.DockerException;
import org.apache.camel.component.docker.producer.AsyncDockerProducer;
import org.apache.camel.component.docker.producer.DockerProducer;
@@ -66,7 +67,16 @@ public class DockerEndpoint extends DefaultEndpoint {
@Override
public Consumer createConsumer(Processor processor) throws Exception {
- throw new UnsupportedOperationException("DockerConsumer is not implemented");
+ DockerOperation operation = configuration.getOperation();
+
+ switch (operation) {
+ case EVENTS:
+ return new DockerEventsConsumer(this, processor);
+ case STATS:
+ return new DockerStatsConsumer(this, processor);
+ default:
+ throw new DockerException(operation + " is not a valid consumer operation");
+ }
}
@Override
http://git-wip-us.apache.org/repos/asf/camel/blob/e27aaedd/components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerEventsConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerEventsConsumer.java b/components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerEventsConsumer.java
new file mode 100644
index 0000000..e87f259
--- /dev/null
+++ b/components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerEventsConsumer.java
@@ -0,0 +1,109 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.docker.consumer;
+
+import java.util.Date;
+
+import com.github.dockerjava.api.command.EventsCmd;
+import com.github.dockerjava.api.model.Event;
+import com.github.dockerjava.core.command.EventsResultCallback;
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.Processor;
+import org.apache.camel.component.docker.DockerClientFactory;
+import org.apache.camel.component.docker.DockerComponent;
+import org.apache.camel.component.docker.DockerConstants;
+import org.apache.camel.component.docker.DockerEndpoint;
+import org.apache.camel.component.docker.DockerHelper;
+import org.apache.camel.impl.DefaultConsumer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class DockerEventsConsumer extends DefaultConsumer {
+ private static final Logger LOG = LoggerFactory.getLogger(DockerEventsConsumer.class);
+
+ private DockerEndpoint endpoint;
+ private DockerComponent component;
+ private EventsCmd eventsCmd;
+
+ public DockerEventsConsumer(DockerEndpoint endpoint, Processor processor) throws Exception {
+ super(endpoint, processor);
+ this.endpoint = endpoint;
+ this.component = (DockerComponent)endpoint.getComponent();
+
+ }
+
+ @Override
+ public DockerEndpoint getEndpoint() {
+ return (DockerEndpoint)super.getEndpoint();
+ }
+
+ /**
+ * Determine the point in time to begin streaming events
+ */
+ private long processInitialEvent() {
+ long currentTime = new Date().getTime();
+ Long initialRange = DockerHelper.getProperty(DockerConstants.DOCKER_INITIAL_RANGE, endpoint.getConfiguration(), null, Long.class);
+ if (initialRange != null) {
+ currentTime = currentTime - initialRange;
+ }
+
+ return currentTime;
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ this.eventsCmd = DockerClientFactory.getDockerClient(component, endpoint.getConfiguration(), null).eventsCmd().withSince(String.valueOf(processInitialEvent()));
+ this.eventsCmd.exec(new EventsCallback());
+
+ super.doStart();
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ this.eventsCmd.close();
+
+ super.doStop();
+ }
+
+ protected class EventsCallback extends EventsResultCallback {
+
+ public void onNext(Event event) {
+ LOG.debug("Received Docker Event: " + event);
+
+ final Exchange exchange = getEndpoint().createExchange();
+ Message message = exchange.getIn();
+ message.setBody(event);
+
+ try {
+ LOG.trace("Processing exchange [{}]...", exchange);
+ getAsyncProcessor().process(exchange, new AsyncCallback() {
+ @Override
+ public void done(boolean doneSync) {
+ LOG.trace("Done processing exchange [{}]...", exchange);
+ }
+ });
+ } catch (Exception e) {
+ exchange.setException(e);
+ }
+ if (exchange.getException() != null) {
+ getExceptionHandler().handleException("Error processing exchange", exchange, exchange.getException());
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e27aaedd/components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerStatsConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerStatsConsumer.java b/components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerStatsConsumer.java
new file mode 100644
index 0000000..015e1ba
--- /dev/null
+++ b/components/camel-docker/src/main/java/org/apache/camel/component/docker/consumer/DockerStatsConsumer.java
@@ -0,0 +1,99 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.docker.consumer;
+
+import com.github.dockerjava.api.command.StatsCmd;
+import com.github.dockerjava.api.model.Statistics;
+import com.github.dockerjava.core.async.ResultCallbackTemplate;
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.Processor;
+import org.apache.camel.component.docker.DockerClientFactory;
+import org.apache.camel.component.docker.DockerComponent;
+import org.apache.camel.component.docker.DockerConstants;
+import org.apache.camel.component.docker.DockerEndpoint;
+import org.apache.camel.component.docker.DockerHelper;
+import org.apache.camel.impl.DefaultConsumer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Docker Consumer for streaming statistical events
+ */
+public class DockerStatsConsumer extends DefaultConsumer {
+
+ private static final transient Logger LOGGER = LoggerFactory.getLogger(DockerStatsConsumer.class);
+
+ private DockerEndpoint endpoint;
+ private DockerComponent component;
+ private StatsCmd statsCmd;
+
+ public DockerStatsConsumer(DockerEndpoint endpoint, Processor processor) throws Exception {
+ super(endpoint, processor);
+ this.endpoint = endpoint;
+ this.component = (DockerComponent)endpoint.getComponent();
+ }
+
+ @Override
+ public DockerEndpoint getEndpoint() {
+ return (DockerEndpoint)super.getEndpoint();
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ String containerId = DockerHelper.getProperty(DockerConstants.DOCKER_CONTAINER_ID, endpoint.getConfiguration(), null, String.class);
+
+ this.statsCmd = DockerClientFactory.getDockerClient(component, endpoint.getConfiguration(), null).statsCmd(containerId);
+ this.statsCmd.exec(new StatsCallback());
+
+ super.doStart();
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ this.statsCmd.close();
+
+ super.doStop();
+ }
+
+ protected class StatsCallback extends ResultCallbackTemplate<StatsCallback, Statistics> {
+
+ public void onNext(Statistics statistics) {
+ LOGGER.debug("Received Docker Statistics Event: " + statistics);
+
+ final Exchange exchange = getEndpoint().createExchange();
+ Message message = exchange.getIn();
+ message.setBody(statistics);
+
+ try {
+ LOGGER.trace("Processing exchange [{}]...", exchange);
+ getAsyncProcessor().process(exchange, new AsyncCallback() {
+ @Override
+ public void done(boolean doneSync) {
+ LOGGER.trace("Done processing exchange [{}]...", exchange);
+ }
+ });
+ } catch (Exception e) {
+ exchange.setException(e);
+ }
+ if (exchange.getException() != null) {
+ getExceptionHandler().handleException("Error processing exchange", exchange, exchange.getException());
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e27aaedd/components/camel-docker/src/test/java/org/apache/camel/component/docker/consumer/DockerEventsConsumerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-docker/src/test/java/org/apache/camel/component/docker/consumer/DockerEventsConsumerTest.java b/components/camel-docker/src/test/java/org/apache/camel/component/docker/consumer/DockerEventsConsumerTest.java
new file mode 100644
index 0000000..8d39e38
--- /dev/null
+++ b/components/camel-docker/src/test/java/org/apache/camel/component/docker/consumer/DockerEventsConsumerTest.java
@@ -0,0 +1,89 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.docker.consumer;
+
+import com.github.dockerjava.api.DockerClient;
+import com.github.dockerjava.api.command.EventsCmd;
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.docker.DockerComponent;
+import org.apache.camel.component.docker.DockerConfiguration;
+import org.apache.camel.component.docker.util.DockerTestUtils;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.modules.junit4.PowerMockRunner;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Consumer test for events on Docker Platform
+ */
+@RunWith(PowerMockRunner.class)
+@PowerMockIgnore({"javax.management.*", "javax.net.ssl.*"})
+public class DockerEventsConsumerTest extends CamelTestSupport {
+ private String host = "localhost";
+ private Integer port = 2375;
+
+ private DockerConfiguration dockerConfiguration;
+
+ @Mock
+ private EventsCmd eventsCmd;
+
+ @Mock
+ private DockerClient dockerClient;
+
+ public void setupMocks() {
+ Mockito.when(dockerClient.eventsCmd()).thenReturn(eventsCmd);
+ Mockito.when(eventsCmd.withSince(Mockito.anyString())).thenReturn(eventsCmd);
+ }
+
+ @Test
+ public void testEvent() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMinimumMessageCount(1);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ public void configure() {
+ from("docker://events?host=" + host + "&port=" + port).log("${body}").to("mock:result");
+ }
+ };
+ }
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext camelContext = super.createCamelContext();
+ dockerConfiguration = new DockerConfiguration();
+ dockerConfiguration.setParameters(DockerTestUtils.getDefaultParameters(host, port, dockerConfiguration));
+
+ DockerComponent dockerComponent = new DockerComponent(dockerConfiguration);
+ dockerComponent.setClient(DockerTestUtils.getClientProfile(host, port, dockerConfiguration), dockerClient);
+ camelContext.addComponent("docker", dockerComponent);
+
+ setupMocks();
+
+ return camelContext;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e27aaedd/components/camel-docker/src/test/java/org/apache/camel/component/docker/consumer/DockerStatsConsumerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-docker/src/test/java/org/apache/camel/component/docker/consumer/DockerStatsConsumerTest.java b/components/camel-docker/src/test/java/org/apache/camel/component/docker/consumer/DockerStatsConsumerTest.java
new file mode 100644
index 0000000..ae020c2
--- /dev/null
+++ b/components/camel-docker/src/test/java/org/apache/camel/component/docker/consumer/DockerStatsConsumerTest.java
@@ -0,0 +1,87 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.docker.consumer;
+
+import com.github.dockerjava.api.DockerClient;
+import com.github.dockerjava.api.command.StatsCmd;
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.docker.DockerComponent;
+import org.apache.camel.component.docker.DockerConfiguration;
+import org.apache.camel.component.docker.util.DockerTestUtils;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+/**
+ * Consumer test for statistics on Docker Platform
+ */
+@RunWith(PowerMockRunner.class)
+@PowerMockIgnore({"javax.management.*", "javax.net.ssl.*"})
+public class DockerStatsConsumerTest extends CamelTestSupport {
+ private String host = "localhost";
+ private Integer port = 2375;
+ private String containerId = "470b9b823e6c";
+
+ private DockerConfiguration dockerConfiguration;
+
+ @Mock
+ private StatsCmd statsCmd;
+
+ @Mock
+ private DockerClient dockerClient;
+
+ public void setupMocks() {
+ Mockito.when(dockerClient.statsCmd(Mockito.anyString())).thenReturn(statsCmd);
+ }
+
+ @Test
+ public void testStats() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMinimumMessageCount(1);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ public void configure() {
+ from("docker://stats?host=" + host + "&port=" + port + "&containerId=" + containerId).log("${body}").to("mock:result");
+ }
+ };
+ }
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext camelContext = super.createCamelContext();
+ dockerConfiguration = new DockerConfiguration();
+ dockerConfiguration.setParameters(DockerTestUtils.getDefaultParameters(host, port, dockerConfiguration));
+
+ DockerComponent dockerComponent = new DockerComponent(dockerConfiguration);
+ dockerComponent.setClient(DockerTestUtils.getClientProfile(host, port, dockerConfiguration), dockerClient);
+ camelContext.addComponent("docker", dockerComponent);
+
+ setupMocks();
+
+ return camelContext;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e27aaedd/components/camel-docker/src/test/java/org/apache/camel/component/docker/util/DockerTestUtils.java
----------------------------------------------------------------------
diff --git a/components/camel-docker/src/test/java/org/apache/camel/component/docker/util/DockerTestUtils.java b/components/camel-docker/src/test/java/org/apache/camel/component/docker/util/DockerTestUtils.java
new file mode 100644
index 0000000..76fe942
--- /dev/null
+++ b/components/camel-docker/src/test/java/org/apache/camel/component/docker/util/DockerTestUtils.java
@@ -0,0 +1,62 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.docker.util;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.camel.component.docker.DockerClientProfile;
+import org.apache.camel.component.docker.DockerConfiguration;
+import org.apache.camel.component.docker.DockerConstants;
+
+public final class DockerTestUtils {
+
+ private DockerTestUtils() {
+
+ }
+
+ public static Map<String, Object> getDefaultParameters(String host, Integer port, DockerConfiguration dockerConfiguration) {
+ Map<String, Object> parameters = new HashMap<String, Object>();
+ parameters.put(DockerConstants.DOCKER_HOST, host);
+ parameters.put(DockerConstants.DOCKER_PORT, port);
+ parameters.put(DockerConstants.DOCKER_EMAIL, dockerConfiguration.getEmail());
+ parameters.put(DockerConstants.DOCKER_SERVER_ADDRESS, dockerConfiguration.getServerAddress());
+ parameters.put(DockerConstants.DOCKER_MAX_PER_ROUTE_CONNECTIONS, dockerConfiguration.getMaxPerRouteConnections());
+ parameters.put(DockerConstants.DOCKER_MAX_TOTAL_CONNECTIONS, dockerConfiguration.getMaxTotalConnections());
+ parameters.put(DockerConstants.DOCKER_SECURE, dockerConfiguration.isSecure());
+ parameters.put(DockerConstants.DOCKER_TLSVERIFY, dockerConfiguration.isTlsVerify());
+ parameters.put(DockerConstants.DOCKER_SOCKET_ENABLED, dockerConfiguration.isSocket());
+
+ return parameters;
+ }
+
+ public static DockerClientProfile getClientProfile(String host, Integer port, DockerConfiguration dockerConfiguration) {
+ DockerClientProfile clientProfile = new DockerClientProfile();
+ clientProfile.setHost(host);
+ clientProfile.setPort(port);
+ clientProfile.setEmail(dockerConfiguration.getEmail());
+ clientProfile.setServerAddress(dockerConfiguration.getServerAddress());
+ clientProfile.setMaxPerRouteConnections(dockerConfiguration.getMaxPerRouteConnections());
+ clientProfile.setMaxTotalConnections(dockerConfiguration.getMaxTotalConnections());
+ clientProfile.setSecure(dockerConfiguration.isSecure());
+ clientProfile.setTlsVerify(dockerConfiguration.isTlsVerify());
+ clientProfile.setSocket(dockerConfiguration.isSocket());
+
+ return clientProfile;
+ }
+
+}
\ No newline at end of file