reorganize python folder

parse json message from planner 
make DRIPCaller closale 

drip-api/src/main/java/nl/uva/sne/drip/api/rest/PlannerController.java

@@ -18,17 +18,13 @@ package nl.uva.sne.drip.api.rest;
 import nl.uva.sne.drip.api.service.SimplePlannerService;
 import nl.uva.sne.drip.commons.types.ToscaRepresentation;
 import java.io.IOException;
-import java.io.UnsupportedEncodingException;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Map;
 import java.util.concurrent.TimeoutException;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import javax.annotation.security.RolesAllowed;
-import nl.uva.sne.drip.commons.types.Message;
-import nl.uva.sne.drip.commons.types.Parameter;
-import nl.uva.sne.drip.commons.utils.Converter;
+import nl.uva.sne.drip.api.exception.NotFoundException;
 import org.json.JSONException;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
@@ -69,6 +65,9 @@ public class PlannerController {
 //            ToscaRepresentation plan = simplePlannerService.getPlan(toscaId);
 //            return plan.getId();
             ToscaRepresentation plan = plannerService.getPlan(toscaId);
+            if (plan == null) {
+                throw new NotFoundException("Could not make plan");
+            }
             return plan.getId();
         } catch (JSONException | IOException | TimeoutException | InterruptedException ex) {
             Logger.getLogger(PlannerController.class.getName()).log(Level.SEVERE, null, ex);

drip-api/src/main/java/nl/uva/sne/drip/api/rest/ProvisionController.java

@@ -79,29 +79,19 @@ public class ProvisionController {
         re.setUserScriptID("58a2112363d41754cca042b4");
         return re;
     }
-        
-    
+
     @RequestMapping(value = "/provision", method = RequestMethod.POST)
     @RolesAllowed({UserService.USER, UserService.ADMIN})
     public @ResponseBody
     String plann(@RequestBody ProvisionRequest req) {
-        DRIPCaller provisioner = null;
-        try {
+
+        try (DRIPCaller provisioner = new ProvisionerCaller(messageBrokerHost);) {
             Message provisionerInvokationMessage = buildProvisionerMessage(req);
-            provisioner = new ProvisionerCaller(messageBrokerHost);
+
             Message response = provisioner.call(provisionerInvokationMessage);
             return "";
         } catch (IOException | TimeoutException | JSONException | InterruptedException ex) {
             Logger.getLogger(ProvisionController.class.getName()).log(Level.SEVERE, null, ex);
-        } finally {
-
-            if (provisioner != null) {
-                try {
-                    provisioner.close();
-                } catch (IOException | TimeoutException ex) {
-                    Logger.getLogger(ProvisionController.class.getName()).log(Level.SEVERE, null, ex);
-                }
-            }
         }
         return null;
     }
@@ -190,7 +180,6 @@ public class ProvisionController {
         return parameters;
     }
 
-    
     private Parameter buildEC2Conf(CloudCredentials cred) throws JsonProcessingException, JSONException, IOException {
 
         Properties prop = Converter.getEC2Properties(cred);

drip-api/src/main/java/nl/uva/sne/drip/api/rpc/DRIPCaller.java

@@ -9,11 +9,17 @@ import com.rabbitmq.client.ConnectionFactory;
 import com.rabbitmq.client.DefaultConsumer;
 import com.rabbitmq.client.Envelope;
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.TimeoutException;
 import nl.uva.sne.drip.commons.types.Message;
+import nl.uva.sne.drip.commons.types.Parameter;
+import org.json.JSONArray;
+import org.json.JSONException;
+import org.json.JSONObject;
 
 /*
  * Copyright 2017 S. Koulouzis, Wang Junchao, Huan Zhou, Yang Hu 
@@ -34,7 +40,7 @@ import nl.uva.sne.drip.commons.types.Message;
  *
  * @author S. Koulouzis
  */
-public abstract class DRIPCaller {
+public abstract class DRIPCaller implements AutoCloseable {
 
     private final Connection connection;
     private final Channel channel;
@@ -76,6 +82,7 @@ public abstract class DRIPCaller {
         return replyQueueName;
     }
 
+    @Override
     public void close() throws IOException, TimeoutException {
         if (channel != null && channel.isOpen()) {
             channel.close();
@@ -85,7 +92,7 @@ public abstract class DRIPCaller {
         }
     }
 
-    public Message call(Message r) throws IOException, TimeoutException, InterruptedException {
+    public Message call(Message r) throws IOException, TimeoutException, InterruptedException, JSONException {
 
         ObjectMapper mapper = new ObjectMapper();
         mapper.configure(JsonParser.Feature.ALLOW_SINGLE_QUOTES, true);
@@ -113,7 +120,24 @@ public abstract class DRIPCaller {
         String strResponse = response.take();
         strResponse = strResponse.replaceAll("'null'", "null").replaceAll("\'", "\"").replaceAll(" ", "");
 //        System.err.println(strResponse);
-        return mapper.readValue(strResponse, Message.class);
+        JSONObject jsonObj = new JSONObject(strResponse);
+        Message responseMessage = new Message();
+
+        responseMessage.setCreationDate((Long) jsonObj.get("creationDate"));
+        JSONArray jsonParams = (JSONArray) jsonObj.get("parameters");
+        List<Parameter> parameters = new ArrayList<>();
+
+        for (int i = 0; i < jsonParams.length(); i++) {
+            JSONObject jsonParam = (JSONObject) jsonParams.get(i);
+
+            Parameter parameter = new Parameter();
+            parameter.setName(jsonParam.getString("name"));
+            parameter.setName(jsonParam.getString("value"));
+            parameters.add(parameter);
+        }
+
+        responseMessage.setParameters(parameters);
+        return responseMessage;//mapper.readValue(strResponse, Message.class);
     }
 
 }

drip-api/src/main/java/nl/uva/sne/drip/api/rpc/PlannerCaller.java

@@ -23,12 +23,12 @@ import java.util.concurrent.TimeoutException;
  *
  * @author S. Koulouzis.
  */
-public class PlannerCaller extends DRIPCaller {
+public class PlannerCaller extends DRIPCaller  {
 
     private static final String REQUEST_QUEUE_NAME = "planner_queue";
 
     public PlannerCaller(String messageBrokerHost) throws IOException, TimeoutException {
-        super(messageBrokerHost,REQUEST_QUEUE_NAME);
+        super(messageBrokerHost, REQUEST_QUEUE_NAME);
     }
 
 }

drip-api/src/main/java/nl/uva/sne/drip/api/service/PlannerService.java

@@ -45,11 +45,22 @@ public class PlannerService {
     private String messageBrokerHost;
 
     public ToscaRepresentation getPlan(String toscaId) throws JSONException, UnsupportedEncodingException, IOException, TimeoutException, InterruptedException {
-        Message plannerInvokationMessage = buildPlannerMessage(toscaId);
-        PlannerCaller planner = new PlannerCaller(messageBrokerHost);
-        Message plannerReturnedMessage = planner.call(plannerInvokationMessage);
 
-        return null;
+        try (PlannerCaller planner = new PlannerCaller(messageBrokerHost)) {
+            Message plannerInvokationMessage = buildPlannerMessage(toscaId);
+
+            Message plannerReturnedMessage = planner.call(plannerInvokationMessage);
+            List<Parameter> parameters = plannerReturnedMessage.getParameters();
+            for(Parameter param: parameters){
+                
+            }
+            ToscaRepresentation tr = new ToscaRepresentation();
+            Map<String, Object> kvMap = null;
+            tr.setKvMap(kvMap);
+            tr.setType(ToscaRepresentation.Type.PLAN);
+
+            return null;
+        }
     }
 
     private Message buildPlannerMessage(String toscaId) throws JSONException, UnsupportedEncodingException {

python/panner/rpc_server.py

+#!/usr/bin/env python
+import pika
+import networkx as nx
+import sys
+import numpy as np
+import sys, argparse
+import operator
+import os
+from toscaparser import *
+from toscaparser.tosca_template import ToscaTemplate
+import re
+import getopt
+from ICPCP import Workflow
+import random
+import time
+import json
+
+
+
+connection = pika.BlockingConnection(pika.ConnectionParameters(host='172.17.0.3'))
+channel = connection.channel()
+channel.queue_declare(queue='planner_queue')
+
+
+def on_request(ch, method, props, body):
+    
+    parsed_message = json.loads(body)
+    value = parsed_message.get('parameters')[0].get('value')
+    parsed_value =  json.loads(value)
+    
+    json1 = parsed_value.get('topology_template').get('node_templates')
+    
+    deadline = 0
+
+    for j in json1:
+        #print json[j]
+        if not json1[j]['type'] == "Switch.nodes.Application.Connection":
+            deadline = int(re.search(r'\d+', json1[j]['properties']['QoS']['response_time']).group())
+
+    #get the nodes from the json
+    nodeDic = {}
+    nodeDic1 = {}
+    i = 1
+    for j in json1:
+        if not json1[j]['type'] == "Switch.nodes.Application.Connection":
+            print j, json1[j]
+        nodeDic[j] = i
+        nodeDic1[i] = j
+        i = i + 1
+
+    #get the links from the json
+    links = []
+    for j in json1:
+        if json1[j]['type'] == "Switch.nodes.Application.Connection":
+            print json1[j]['properties']['source']['component_name']
+            print json1[j]['properties']['target']['component_name']
+            link= {}
+            link['source'] = nodeDic[json1[j]['properties']['source']['component_name']]
+            link['target'] = nodeDic[json1[j]['properties']['target']['component_name']]
+            link['weight'] = random.randint(1, 10)
+            links.append(link)
+
+    # compose the json as input of the workflow
+    wfJson = {}
+    wfJson['workflow'] = {}
+
+    nodesList = []
+    sorted_nodeDic = sorted(nodeDic.items(), key=operator.itemgetter(1))
+    for key, value in sorted_nodeDic:
+        v = {}
+        v['name'] = value
+        nodesList.append(v)
+    wfJson['workflow']['nodes'] = nodesList
+    wfJson['workflow']['links'] = links
+    #print deadline
+    
+    wfJson['price'] = "5,2,1"
+    wfJson['deadline'] = {'2': deadline}
+
+    #generate performance
+    performance = {}
+    for key, value in sorted_nodeDic:
+        performance[str(value)] = "1,2,3"
+    wfJson['performance'] = performance
+    
+    print wfJson
+
+    #send request to the server
+    start = time.time()
+    wf = Workflow()
+    wf.init(wfJson)
+    wf.ic_pcp()
+    #print content['workflow']
+    #return 
+    res = wf.generateJSON()
+    end = time.time()
+    print (end - start)
+    
+    # generate the json files in the corresponding format as the 
+    outcontent = {}
+    current_milli_time = lambda: int(round(time.time() * 1000))
+    outcontent["creationDate"] = current_milli_time()
+    outcontent["parameters"] = []
+    par1 = {}
+    par1["url"] = 'null'
+    par1["encoding"] = "UTF-8"
+    par1["name"] = "plan"
+    par1["value"] = res
+    par1["attributes"] = 'null'
+    outcontent["parameters"].append(par1)
+    
+    response = outcontent
+    print(response)
+    ch.basic_publish(exchange='',
+                     routing_key=props.reply_to,
+                     properties=pika.BasicProperties(correlation_id = \
+                                                         props.correlation_id),
+                     body=str(response))
+    ch.basic_ack(delivery_tag = method.delivery_tag)
+
+channel.basic_qos(prefetch_count=1)
+channel.basic_consume(on_request, queue='planner_queue')
+
+print(" [x] Awaiting RPC requests")
+channel.start_consuming()
\ No newline at end of file

python/rpc_server.py

 #!/usr/bin/env python
 import pika
-import networkx as nx
-import sys
-import numpy as np
-import sys, argparse
-import operator
-import os
-from toscaparser import *
-from toscaparser.tosca_template import ToscaTemplate
-import re
-import getopt
-from ICPCP import Workflow
 import random
 import time
 import json
-import time
 
 
 
@@ -24,92 +12,11 @@ channel.queue_declare(queue='planner_queue')
 
 
 def on_request(ch, method, props, body):
-    
     parsed_message = json.loads(body)
-    value = parsed_message.get('parameters')[0].get('value')
-    parsed_value =  json.loads(value)
-    
-    json1 = parsed_value.get('topology_template').get('node_templates')
-    
-    deadline = 0
-
-    for j in json1:
-        #print json[j]
-        if not json1[j]['type'] == "Switch.nodes.Application.Connection":
-            deadline = int(re.search(r'\d+', json1[j]['properties']['QoS']['response_time']).group())
-
-    #get the nodes from the json
-    nodeDic = {}
-    nodeDic1 = {}
-    i = 1
-    for j in json1:
-        if not json1[j]['type'] == "Switch.nodes.Application.Connection":
-            print j, json1[j]
-        nodeDic[j] = i
-        nodeDic1[i] = j
-        i = i + 1
-
-    #get the links from the json
-    links = []
-    for j in json1:
-        if json1[j]['type'] == "Switch.nodes.Application.Connection":
-            print json1[j]['properties']['source']['component_name']
-            print json1[j]['properties']['target']['component_name']
-            link= {}
-            link['source'] = nodeDic[json1[j]['properties']['source']['component_name']]
-            link['target'] = nodeDic[json1[j]['properties']['target']['component_name']]
-            link['weight'] = random.randint(1, 10)
-            links.append(link)
-
-    # compose the json as input of the workflow
-    wfJson = {}
-    wfJson['workflow'] = {}
-
-    nodesList = []
-    sorted_nodeDic = sorted(nodeDic.items(), key=operator.itemgetter(1))
-    for key, value in sorted_nodeDic:
-        v = {}
-        v['name'] = value
-        nodesList.append(v)
-    wfJson['workflow']['nodes'] = nodesList
-    wfJson['workflow']['links'] = links
-    #print deadline
     
-    wfJson['price'] = "5,2,1"
-    wfJson['deadline'] = {'2': deadline}
-
-    #generate performance
-    performance = {}
-    for key, value in sorted_nodeDic:
-        performance[str(value)] = "1,2,3"
-    wfJson['performance'] = performance
-    
-    print wfJson
-
-    #send request to the server
-    start = time.time()
-    wf = Workflow()
-    wf.init(wfJson)
-    wf.ic_pcp()
-    #print content['workflow']
-    #return 
-    res = wf.generateJSON()
-    end = time.time()
-    print (end - start)
+    response = consume_message(parsed_message)
     
-    # generate the json files in the corresponding format as the 
-    outcontent = {}
-    outcontent["creationDate"] = time.time()
-    outcontent["parameters"] = []
-    par1 = {}
-    par1["url"] = 'null'
-    par1["encoding"] = "UTF-8"
-    par1["value"] = str(res) 
-    par1["attributes"] = 'null'
-    outcontent["parameters"].append(par1)
     
-    response = outcontent
-    print(response)
     ch.basic_publish(exchange='',
                      routing_key=props.reply_to,
                      properties=pika.BasicProperties(correlation_id = \