Initial commit

Author: Carlos León

Dockerfile

@@ -0,0 +1,6 @@
+FROM java:openjdk-8-alpine
+
+WORKDIR /usr/src/app
+COPY ./target/*.jar ./app.jar
+
+ENTRYPOINT ["java","-Djava.security.egd=file:/dev/urandom","-jar","./app.jar", "--port=80"]

pom.xml

@@ -0,0 +1,32 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xmlns="http://maven.apache.org/POM/4.0.0"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>works.weave.socks</groupId>
+    <artifactId>queue-master</artifactId>
+    <packaging>jar</packaging>
+
+    <name>queue-master</name>
+    <description>Shipping queue master for socks eCommerce application
+    </description>
+
+    <parent>
+        <groupId>works.weave</groupId>
+        <artifactId>socks</artifactId>
+        <version>0.0.1-SNAPSHOT</version>
+    </parent>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-amqp</artifactId>
+        </dependency>
+        <dependency>
+          <groupId>com.github.docker-java</groupId>
+          <artifactId>docker-java</artifactId>
+          <version>3.0.0</version>
+        </dependency>
+    </dependencies>
+</project>

src/main/java/works/weave/socks/DockerSpawner.java

@@ -0,0 +1,72 @@
+package works.weave.socks;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.github.dockerjava.api.DockerClient;
+import com.github.dockerjava.api.command.CreateContainerResponse;
+import com.github.dockerjava.api.model.Network;
+import com.github.dockerjava.core.DockerClientBuilder;
+import com.github.dockerjava.core.DockerClientConfig;
+import com.github.dockerjava.core.command.PullImageResultCallback;
+import com.github.dockerjava.core.command.ExecStartResultCallback;
+import com.github.dockerjava.api.exception.DockerException;
+import org.springframework.stereotype.Component;
+
+import java.lang.Exception;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.List;
+
+@Component
+public class DockerSpawner {
+    private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+	private DockerClient dc;
+	private ExecutorService dockerPool;
+
+	private String imageName = "weaveworksdemos/worker";
+	private String imageVersion = "latest";
+	private String networkId = "weavedemo_backoffice";
+	private int poolSize = 50;
+
+	public void init() {
+		if (dc == null) {
+			DockerClientConfig.DockerClientConfigBuilder builder = DockerClientConfig.createDefaultConfigBuilder();
+
+            DockerClientConfig config = builder.build();
+            dc = DockerClientBuilder.getInstance(config).build();
+
+            dc.pullImageCmd(imageName).withTag(imageVersion).exec(new PullImageResultCallback()).awaitSuccess();
+		}
+		if (dockerPool == null) {
+			dockerPool = Executors.newFixedThreadPool(poolSize);
+		}
+	}
+
+	public void spawn() {
+		dockerPool.execute(new Runnable() {
+		    public void run() {
+				logger.info("Spawning new container");
+				try {
+					CreateContainerResponse container = dc.createContainerCmd(imageName + ":" + imageVersion).withNetworkMode(networkId).withCmd("ping", "rabbitmq").exec();
+					String containerId = container.getId();
+					dc.startContainerCmd(containerId).exec();
+					logger.info("Spawned container with id: " + container.getId() + " on network: " + networkId);
+					// TODO instead of just sleeping, call await on the container and remove once it's completed.
+					Thread.sleep(40000);
+					try {
+						dc.stopContainerCmd(containerId).exec();
+					}
+					catch (DockerException e) {
+						logger.info("Container already stopped. (This is expected).");
+					}
+					dc.removeContainerCmd(containerId).exec();
+					logger.info("Removed Container:" + containerId);
+				} catch (Exception e) {
+					logger.error("Exception trying to launch/remove worker container. " + e);
+				}
+		    }
+		});
+	}
+}

src/main/java/works/weave/socks/QueueMasterApplication.java

@@ -0,0 +1,45 @@
+package works.weave.socks;
+
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.core.TopicExchange;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.CommandLineRunner;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.Bean;
+
+@SpringBootApplication
+public class QueueMasterApplication implements CommandLineRunner {
+
+	final static String queueName = "shipping-task";
+
+	@Autowired
+	RabbitTemplate rabbitTemplate;
+
+	public static void main(String[] args) throws InterruptedException {
+		SpringApplication.run(QueueMasterApplication.class, args);
+	}
+
+	@Bean
+	Queue queue() {
+		return new Queue(queueName, false);
+	}
+
+	@Bean
+	TopicExchange exchange() {
+		return new TopicExchange("shipping-task-exchange");
+	}
+
+	@Bean
+	Binding binding(Queue queue, TopicExchange exchange) {
+		return BindingBuilder.bind(queue).to(exchange).with(queueName);
+	}
+
+    @Override
+    public void run(String... args) throws Exception {
+        System.out.println("Starting QueueMasterApplication...");
+    }
+}

src/main/java/works/weave/socks/RabbitMqConfiguration.java

@@ -0,0 +1,47 @@
+package works.weave.socks;
+
+import org.springframework.amqp.core.AmqpAdmin;
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
+import org.springframework.amqp.support.converter.DefaultClassMapper;
+import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
+import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class RabbitMqConfiguration
+{
+    @Bean
+    public ConnectionFactory connectionFactory()
+    {
+        CachingConnectionFactory connectionFactory = new CachingConnectionFactory("rabbitmq");
+        // CachingConnectionFactory connectionFactory = new CachingConnectionFactory("192.168.99.104");
+        connectionFactory.setUsername("guest");
+        connectionFactory.setPassword("guest");
+        return connectionFactory;
+    }
+
+    @Bean
+    public AmqpAdmin amqpAdmin()
+    {
+        return new RabbitAdmin(connectionFactory());
+    }
+
+    @Bean
+    public MessageConverter jsonMessageConverter()
+    {
+        final Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter();
+        converter.setClassMapper(classMapper());
+        return converter;
+    }
+
+    @Bean
+    public DefaultClassMapper classMapper()
+    {
+        DefaultClassMapper typeMapper = new DefaultClassMapper();
+        typeMapper.setDefaultType(Shipment.class);
+        return typeMapper;
+    }
+}

src/main/java/works/weave/socks/Shipment.java

@@ -0,0 +1,22 @@
+package works.weave.socks;
+
+public class Shipment {
+	private String id;
+	private String name;
+
+	public String getId() {
+		return id;
+	}
+
+	public void setId(String id) {
+		this.id = id;
+	}
+
+	public String getName() {
+		return name;
+	}
+
+	public void setName(String name) {
+		this.name = name;
+	}
+}

src/main/java/works/weave/socks/ShippingConsumerConfiguration.java

@@ -0,0 +1,46 @@
+package works.weave.socks;
+
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class ShippingConsumerConfiguration extends RabbitMqConfiguration
+{
+	protected final String queueName = "shipping-task";
+
+    @Autowired
+    private ShippingTaskHandler shippingTaskHandler;
+
+	@Bean
+	public RabbitTemplate rabbitTemplate() {
+		RabbitTemplate template = new RabbitTemplate(connectionFactory());
+		template.setQueue(this.queueName);
+        template.setMessageConverter(jsonMessageConverter());
+		return template;
+	}
+
+    @Bean
+	public Queue queueName() {
+		return new Queue(this.queueName, false);
+	}
+
+	@Bean
+	public SimpleMessageListenerContainer listenerContainer() {
+		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
+		container.setConnectionFactory(connectionFactory());
+		container.setQueueNames(this.queueName);
+		container.setMessageListener(messageListenerAdapter());
+
+		return container;
+	}
+
+    @Bean
+    public MessageListenerAdapter messageListenerAdapter() {
+        return new MessageListenerAdapter(shippingTaskHandler, jsonMessageConverter());
+    }
+}

src/main/java/works/weave/socks/ShippingTaskHandler.java

@@ -0,0 +1,17 @@
+package works.weave.socks;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+@Component
+public class ShippingTaskHandler {
+
+	@Autowired
+	DockerSpawner docker;
+
+	public void handleMessage(Shipment shipment) {
+		System.out.println("Received shipment task: " + shipment.getName());
+		docker.init();
+		docker.spawn();
+	}
+}

src/main/resources/log4j.properties

@@ -0,0 +1,11 @@
+# Set root logger level to DEBUG and its only appender to A1.
+log4j.rootLogger=WARN, A1
+
+# A1 is set to be a ConsoleAppender.
+log4j.appender.A1=org.apache.log4j.ConsoleAppender
+
+# A1 uses PatternLayout.
+log4j.appender.A1.layout=org.apache.log4j.PatternLayout
+log4j.appender.A1.layout.ConversionPattern=%-4r [%t] %-5p %c %x - %m%n
+
+log4j.category.org.springframework=INFO