Pass file as bytes

nb-configuration.xml

@@ -6,7 +6,6 @@ The configuration is intended to be shared among all the users of project and
 therefore it is assumed to be part of version control checkout.
 Without this configuration present, some functionality in the IDE may be limited or fail altogether.
 -->
-    <libraries xmlns="http://www.netbeans.org/ns/cdnjs-libraries/1"/>
     <properties xmlns="http://www.netbeans.org/ns/maven-properties-data/1">
         <!--
 Properties that influence various parts of the IDE, especially code formatting and the like. 
@@ -26,4 +25,5 @@ Any value defined here will override the pom.xml file value but is only applicab
         <org-netbeans-modules-whitelist.whitelist-oracle>false</org-netbeans-modules-whitelist.whitelist-oracle>
         <org-netbeans-modules-web-clientproject-api.js_2e_libs_2e_folder>js/libs</org-netbeans-modules-web-clientproject-api.js_2e_libs_2e_folder>
     </properties>
+    <libraries xmlns="http://www.netbeans.org/ns/cdnjs-libraries/1"/>
 </project-shared-configuration>

pom.xml

@@ -33,11 +33,11 @@
             <type>jar</type>
         </dependency>
         <dependency>
-            <groupId>org.springframework.amqp</groupId>
-            <artifactId>spring-rabbit</artifactId>
-            <version>1.6.6.RELEASE</version>
-            <type>jar</type>
+            <groupId>com.rabbitmq</groupId>
+            <artifactId>amqp-client</artifactId>
+            <version>4.0.2</version>
         </dependency>
+        
         <dependency>
             <groupId>org.springframework.data</groupId>
             <artifactId>spring-data-mongodb</artifactId>
@@ -86,8 +86,8 @@
                 <artifactId>maven-compiler-plugin</artifactId>
                 <version>2.3.2</version>
                 <configuration>
-                    <source>1.6</source>
-                    <target>1.6</target>
+                    <source>1.7</source>
+                    <target>1.7</target>
                     <compilerArguments>
                         <endorseddirs>${endorsed.dir}</endorseddirs>
                     </compilerArguments>

src/main/java/nl/uva/sne/drip/rest/UploadToscaController.java

@@ -15,13 +15,9 @@
  */
 package nl.uva.sne.drip.rest;
 
-import nl.uva.sne.drip.rpc.Panner;
-import java.io.BufferedOutputStream;
+import nl.uva.sne.drip.rpc.Planner;
 import java.io.File;
-import java.io.FileOutputStream;
 import java.io.IOException;
-import java.nio.file.Files;
-import java.util.UUID;
 import java.util.concurrent.TimeoutException;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -56,21 +52,17 @@ public class UploadToscaController {
         if (!file.isEmpty()) {
             try {
 
-                System.err.println(messageBrokerHost);
-
                 String originalFileName = file.getOriginalFilename();
                 String name = originalFileName + System.currentTimeMillis();
                 File targetToscaFile = new File(inputToscaFolderPath + File.separator + name);
                 file.transferTo(targetToscaFile);
 
-                Panner planner = new Panner();
-                planner.plan(targetToscaFile);
+                Planner planner = new Planner(messageBrokerHost);
+                String returned = planner.plan(targetToscaFile);
+                System.err.println(returned);
+                planner.close();
                 return "You successfully uploaded " + name + " into " + name + "-uploaded !";
-            } catch (IOException ex) {
-                return "Upload failed. " + ex.getMessage();
-            } catch (IllegalStateException ex) {
-                return "Upload failed. " + ex.getMessage();
-            } catch (TimeoutException ex) {
+            } catch (IOException | IllegalStateException | TimeoutException | InterruptedException ex) {
                 Logger.getLogger(UploadToscaController.class.getName()).log(Level.SEVERE, null, ex);
             }
         }

src/main/java/nl/uva/sne/drip/rpc/DRIPComponent.java

@@ -20,27 +20,17 @@ import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
 import java.io.IOException;
 import java.util.concurrent.TimeoutException;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.stereotype.Component;
-import org.springframework.web.bind.annotation.RestController;
 
 /**
  *
  * @author S. Koulouzis.
  */
-@Component
-@RestController
 class DRIPComponent {
-    
-    @Autowired
-    @Value("${message.broker.host}")
-    private String messageBrokerHost;
 
     private Connection connection;
     private Channel channel;
 
-    private void connect() throws IOException, TimeoutException {
+    private void connect(String messageBrokerHost) throws IOException, TimeoutException {
         ConnectionFactory factory = new ConnectionFactory();
         factory.setHost(messageBrokerHost);
 
@@ -55,9 +45,9 @@ class DRIPComponent {
     /**
      * @return the channel
      */
-    public Channel getChannel() throws IOException, TimeoutException {
+    public Channel getChannel(String messageBrokerHost) throws IOException, TimeoutException {
         if (channel == null) {
-            connect();
+            connect(messageBrokerHost);
         }
         return channel;
     }

src/main/java/nl/uva/sne/drip/rpc/Panner.java → src/main/java/nl/uva/sne/drip/rpc/Planner.java

@@ -15,40 +15,64 @@
  */
 package nl.uva.sne.drip.rpc;
 
+import java.io.File;
+import java.io.IOException;
+
 import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
 import com.rabbitmq.client.DefaultConsumer;
 import com.rabbitmq.client.Envelope;
-import java.io.File;
-import java.io.IOException;
+import com.rabbitmq.client.RpcClient;
+import java.nio.file.Files;
+import java.nio.file.Paths;
 import java.util.UUID;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.TimeoutException;
-import org.springframework.stereotype.Component;
 
 /**
  *
  * @author S. Koulouzis.
  */
-@Component
-public class Panner extends DRIPComponent {
+public class Planner {
 
-    private String queueName = "planner_queue";
+    private static final String REQUEST_QUEUE_NAME = "planner_queue";
+    private final Connection connection;
+    private final Channel channel;
+    private final String replyQueueName;
 
-    public void plan(File inputToscaFile) throws IOException, TimeoutException {
-        queueName = getChannel().queueDeclare().getQueue();
-        final String corrId = UUID.randomUUID().toString();
+    public Planner(String messageBrokerHost) throws IOException, TimeoutException {
+
+        ConnectionFactory factory = new ConnectionFactory();
+        factory.setHost(messageBrokerHost);
+        factory.setHost(messageBrokerHost);
+        factory.setPort(AMQP.PROTOCOL.PORT);
+        //factory.setUsername("guest");
+        //factory.setPassword("pass");
 
+        connection = factory.newConnection();
+        channel = connection.createChannel();
+        // create a single callback queue per client not per requests. 
+        replyQueueName = channel.queueDeclare().getQueue();
+    }
+
+    public String plan(File inputToscaFile) throws IOException, TimeoutException, InterruptedException {
+
+        //Build a correlation ID to distinguish responds 
+        final String corrId = UUID.randomUUID().toString();
         AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                 .correlationId(corrId)
-                .replyTo(queueName)
+                .replyTo(replyQueueName)
                 .build();
 
-        getChannel().basicPublish("", queueName, props, inputToscaFile.getName().getBytes("UTF-8"));
+        byte[] fileContentsBytes = Files.readAllBytes(Paths.get(inputToscaFile.getAbsolutePath()));
+        channel.basicPublish("", REQUEST_QUEUE_NAME, props, fileContentsBytes);
 
         final BlockingQueue<String> response = new ArrayBlockingQueue(1);
 
-        getChannel().basicConsume(queueName, true, new DefaultConsumer(getChannel()) {
+        channel.basicConsume(replyQueueName, true, new DefaultConsumer(channel) {
             @Override
             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                 if (properties.getCorrelationId().equals(corrId)) {
@@ -56,8 +80,17 @@ public class Panner extends DRIPComponent {
                 }
             }
         });
-
-//        return response.take();
+//        channel.close();
+//        connection.close();
+        return response.take();
     }
 
+    public void close() throws IOException, TimeoutException {
+        if (channel != null && channel.isOpen()) {
+            channel.close();
+        }
+        if (connection != null && connection.isOpen()) {
+            connection.close();
+        }
+    }
 }