@@ -0,0 +1,1646 @@
+package com.tito.enigma.ipc;%0D%0A%0D%0Aimport java.io.IOException;%0D%0Aimport java.net.ServerSocket;%0D%0Aimport java.net.Socket;%0D%0Aimport java.util.ArrayList;%0D%0Aimport java.util.List;%0D%0A%0D%0Aimport org.apache.commons.cli.CommandLine;%0D%0Aimport org.apache.commons.cli.Options;%0D%0Aimport org.apache.commons.logging.Log;%0D%0Aimport org.apache.commons.logging.LogFactory;%0D%0A%0D%0Aimport com.tito.enigma.yarn.task.Tasklet;%0D%0A%0D%0Apublic class QueueEngine extends Tasklet %7B%0D%0A%09private static final Log LOG = LogFactory.getLog(QueueEngine.class);%0D%0A%0D%0A%09Queue queue = new Queue();%0D%0A%09private int port;%0D%0A%0D%0A%09@Override%0D%0A%09public boolean init(CommandLine commandLine) %7B%0D%0A%09%09if (!commandLine.hasOption(%22queueEnginePort%22)) %7B%0D%0A%09%09%09LOG.error(%22Missing Port for QueueEngine%22);%0D%0A%09%09%09return false;%0D%0A%09%09%7D%0D%0A%09%09port = Integer.parseInt(commandLine.getOptionValue(%22queueEnginePort%22));%0D%0A%09%09return true;%0D%0A%0D%0A%09%7D%0D%0A%0D%0A%09@Override%0D%0A%09public void setupOptions(Options opts) %7B%0D%0A%09%09opts.addOption(%22queueEnginePort%22, true, %22port to listen to%22);%0D%0A%0D%0A%09%7D%0D%0A%0D%0A%09@Override%0D%0A%09public boolean start() %7B%0D%0A%09%09List%3CThread%3E clientHandlers = new ArrayList%3C%3E();%0D%0A%0D%0A%09%09ServerSocket serverSocket = null;%0D%0A%09%09try %7B%0D%0A%09%09%09serverSocket = new ServerSocket(port);%0D%0A%09%09%09while (true) %7B%0D%0A%09%09%09%09Socket clientSocket = serverSocket.accept();%0D%0A%09%09%09%09Thread clientHandler = new Thread(new ClientHandler(clientSocket));%0D%0A%09%09%09%09clientHandlers.add(clientHandler);%0D%0A%09%09%09%09clientHandler.start();%0D%0A%09%09%09%7D%0D%0A%09%09%7D catch (IOException e) %7B%0D%0A%09%09%09e.printStackTrace();%0D%0A%09%09%7D finally %7B%0D%0A%09%09%09try %7B%0D%0A%09%09%09%09if (serverSocket != null) %7B%0D%0A%09%09%09%09%09serverSocket.close();%0D%0A%09%09%09%09%7D%0D%0A%09%09%09%7D catch (IOException e) %7B%0D%0A%09%09%09%09// TODO Auto-generated catch block%0D%0A%09%09%09%09e.printStackTrace();%0D%0A%09%09%09%7D%0D%0A%09%09%7D%0D%0A%09%09return false;%0D%0A%09%7D%0D%0A%0D%0A%7D%0D%0A

update queueEngine