@Test public void onMessageTest() throws Exception { VMDAO vmDAO = mock(VMDAO.class); Application application = new Application(); application.setName("SUPER_APP"); Deployment deployment = new Deployment(); deployment.setApplication(application); deployment.setId(22); eu.ascetic.paas.applicationmanager.model.VM vm = new eu.ascetic.paas.applicationmanager.model.VM(); vm.setProviderId("providerId"); vm.setId(111); vm.setProviderVmId("489c0769-2ad3-4f76-a5d0-dedff0877f09"); vm.setDeployment(deployment); vm.setCpuActual(2); vm.setRamActual(1024); vm.setSwapActual(122); vm.setDiskActual(23); vm.setPriceSchema(2l); when(vmDAO.getVMWithProviderVMId("489c0769-2ad3-4f76-a5d0-dedff0877f09", "providerId")).thenReturn(vm); MetricsListener metricsListener = new MetricsListener("amqp: "vm.*.item.power" , "providerId", vmDAO); AmqpMessageReceiver receiver = new AmqpMessageReceiver(Configuration.amqpAddress, Configuration.amqpUsername, Configuration.amqpPassword, "APPLICATION.SUPER_APP.DEPLOYMENT.22.VM.111.METRIC.power", true); AmqpListListener listener = new AmqpListListener(); receiver.setMessageConsumer(listener); String json = "{\n" + " \"name\" : \"power\",\n" + " \"value\" : 0.1,\n" + " \"units\" : \"units\",\n" + " \"timestamp\" : 22\n" + "}"; AmqpMessageProducer producer = new AmqpMessageProducer(Configuration.amqpAddress, Configuration.amqpUsername, Configuration.amqpPassword, "vm.489c0769-2ad3-4f76-a5d0-dedff0877f09.item.power", true); producer.sendMessage(json); producer.close(); Thread.sleep(2000l); assertEquals(1, listener.getTextMessages().size()); assertEquals("APPLICATION.SUPER_APP.DEPLOYMENT.22.VM.111.METRIC.power", listener.getTextMessages().get(0).getJMSDestination().toString()); ApplicationManagerMessage message = ModelConverter.jsonToApplicationManagerMessage(listener.getTextMessages().get(0).getText()); assertEquals("SUPER_APP", message.getApplicationId()); assertEquals("22", message.getDeploymentId()); assertEquals("111", message.getVms().get(0).getVmId()); assertEquals("power", message.getVms().get(0).getMetricName()); assertEquals("489c0769-2ad3-4f76-a5d0-dedff0877f09", message.getVms().get(0).getIaasVmId()); assertEquals("providerId", message.getVms().get(0).getProviderId()); assertEquals(0.1, message.getVms().get(0).getValue(), 0.0001); assertEquals("units", message.getVms().get(0).getUnits()); assertEquals(22l, message.getVms().get(0).getTimestamp()); assertEquals(2, message.getVms().get(0).getCpu()); assertEquals(1024, message.getVms().get(0).getRam()); assertEquals(122, message.getVms().get(0).getSwap()); assertEquals(2, message.getVms().get(0).getPriceSchema()); assertEquals(23, message.getVms().get(0).getDisk()); producer = new AmqpMessageProducer(Configuration.amqpAddress, Configuration.amqpUsername, Configuration.amqpPassword, "vm.489c0769-2ad3-4f76-a5d0-dedff0877f0.item.power", true); producer.sendMessage(json); producer.close(); Thread.sleep(2000l); assertEquals(1, listener.getTextMessages().size()); receiver.close(); metricsListener.close(); }
@Override public void onMessage(Message message) { try { TextMessage textMessage = (TextMessage) message; String destination = textMessage.getJMSDestination().toString(); logger.debug("Message received for destionation: " + destination); String text = (String) textMessage.getText(); logger.debug("Text Message: " + text); IaaSMessage im = Converter.iaasMessageFromJSONToObject(text); logger.debug("DESTINATION: " + destination); String vmID = destination.split("\\.")[1]; logger.debug("VMDAO: " + vmDAO + " vmID " + vmID); eu.ascetic.paas.applicationmanager.model.VM vmFromDB = vmDAO.getVMWithProviderVMId(vmID, providerId); if(vmFromDB != null) { String topicToSend = "APPLICATION." + vmFromDB.getDeployment().getApplication().getName() + ".DEPLOYMENT." + vmFromDB.getDeployment().getId() + ".VM." + vmFromDB.getId() + ".METRIC." + im.getName(); VM vm = new VM(); vm.setIaasVmId(vmID); vm.setMetricName(im.getName()); vm.setUnits(im.getUnits()); vm.setValue(im.getValue()); vm.setTimestamp(im.getTimestamp()); vm.setVmId("" + vmFromDB.getId()); vm.setProviderId(providerId); vm.setCpu(vmFromDB.getCpuActual()); vm.setRam(vmFromDB.getRamActual()); vm.setDisk(vmFromDB.getDiskActual()); vm.setSwap(vmFromDB.getSwapActual()); if(vmFromDB.getPriceSchema() != null) { vm.setPriceSchema(vmFromDB.getPriceSchema().intValue()); } ApplicationManagerMessage amM = new ApplicationManagerMessage(); amM.setApplicationId(vmFromDB.getDeployment().getApplication().getName()); amM.setDeploymentId("" + vmFromDB.getDeployment().getId()); amM.addVM(vm); logger.debug("Message to be sent: " + ModelConverter.applicationManagerMessageToJSON(amM)); AmqpProducer.sendMessage(topicToSend, amM, false); } } catch (JMSException e) { logger.info("Error receiving message from: " + host + " to the topic: " + topic); logger.info(e.getMessage()); } }
MetricsListener implements MessageListener { @Override public void onMessage(Message message) { try { TextMessage textMessage = (TextMessage) message; String destination = textMessage.getJMSDestination().toString(); logger.debug("Message received for destionation: " + destination); String text = (String) textMessage.getText(); logger.debug("Text Message: " + text); IaaSMessage im = Converter.iaasMessageFromJSONToObject(text); logger.debug("DESTINATION: " + destination); String vmID = destination.split("\\.")[1]; logger.debug("VMDAO: " + vmDAO + " vmID " + vmID); eu.ascetic.paas.applicationmanager.model.VM vmFromDB = vmDAO.getVMWithProviderVMId(vmID, providerId); if(vmFromDB != null) { String topicToSend = "APPLICATION." + vmFromDB.getDeployment().getApplication().getName() + ".DEPLOYMENT." + vmFromDB.getDeployment().getId() + ".VM." + vmFromDB.getId() + ".METRIC." + im.getName(); VM vm = new VM(); vm.setIaasVmId(vmID); vm.setMetricName(im.getName()); vm.setUnits(im.getUnits()); vm.setValue(im.getValue()); vm.setTimestamp(im.getTimestamp()); vm.setVmId("" + vmFromDB.getId()); vm.setProviderId(providerId); vm.setCpu(vmFromDB.getCpuActual()); vm.setRam(vmFromDB.getRamActual()); vm.setDisk(vmFromDB.getDiskActual()); vm.setSwap(vmFromDB.getSwapActual()); if(vmFromDB.getPriceSchema() != null) { vm.setPriceSchema(vmFromDB.getPriceSchema().intValue()); } ApplicationManagerMessage amM = new ApplicationManagerMessage(); amM.setApplicationId(vmFromDB.getDeployment().getApplication().getName()); amM.setDeploymentId("" + vmFromDB.getDeployment().getId()); amM.addVM(vm); logger.debug("Message to be sent: " + ModelConverter.applicationManagerMessageToJSON(amM)); AmqpProducer.sendMessage(topicToSend, amM, false); } } catch (JMSException e) { logger.info("Error receiving message from: " + host + " to the topic: " + topic); logger.info(e.getMessage()); } } }
MetricsListener implements MessageListener { @Override public void onMessage(Message message) { try { TextMessage textMessage = (TextMessage) message; String destination = textMessage.getJMSDestination().toString(); logger.debug("Message received for destionation: " + destination); String text = (String) textMessage.getText(); logger.debug("Text Message: " + text); IaaSMessage im = Converter.iaasMessageFromJSONToObject(text); logger.debug("DESTINATION: " + destination); String vmID = destination.split("\\.")[1]; logger.debug("VMDAO: " + vmDAO + " vmID " + vmID); eu.ascetic.paas.applicationmanager.model.VM vmFromDB = vmDAO.getVMWithProviderVMId(vmID, providerId); if(vmFromDB != null) { String topicToSend = "APPLICATION." + vmFromDB.getDeployment().getApplication().getName() + ".DEPLOYMENT." + vmFromDB.getDeployment().getId() + ".VM." + vmFromDB.getId() + ".METRIC." + im.getName(); VM vm = new VM(); vm.setIaasVmId(vmID); vm.setMetricName(im.getName()); vm.setUnits(im.getUnits()); vm.setValue(im.getValue()); vm.setTimestamp(im.getTimestamp()); vm.setVmId("" + vmFromDB.getId()); vm.setProviderId(providerId); vm.setCpu(vmFromDB.getCpuActual()); vm.setRam(vmFromDB.getRamActual()); vm.setDisk(vmFromDB.getDiskActual()); vm.setSwap(vmFromDB.getSwapActual()); if(vmFromDB.getPriceSchema() != null) { vm.setPriceSchema(vmFromDB.getPriceSchema().intValue()); } ApplicationManagerMessage amM = new ApplicationManagerMessage(); amM.setApplicationId(vmFromDB.getDeployment().getApplication().getName()); amM.setDeploymentId("" + vmFromDB.getDeployment().getId()); amM.addVM(vm); logger.debug("Message to be sent: " + ModelConverter.applicationManagerMessageToJSON(amM)); AmqpProducer.sendMessage(topicToSend, amM, false); } } catch (JMSException e) { logger.info("Error receiving message from: " + host + " to the topic: " + topic); logger.info(e.getMessage()); } } MetricsListener(String amqpUrl, String topic, String providerId, VMDAO vmDAO); }
MetricsListener implements MessageListener { @Override public void onMessage(Message message) { try { TextMessage textMessage = (TextMessage) message; String destination = textMessage.getJMSDestination().toString(); logger.debug("Message received for destionation: " + destination); String text = (String) textMessage.getText(); logger.debug("Text Message: " + text); IaaSMessage im = Converter.iaasMessageFromJSONToObject(text); logger.debug("DESTINATION: " + destination); String vmID = destination.split("\\.")[1]; logger.debug("VMDAO: " + vmDAO + " vmID " + vmID); eu.ascetic.paas.applicationmanager.model.VM vmFromDB = vmDAO.getVMWithProviderVMId(vmID, providerId); if(vmFromDB != null) { String topicToSend = "APPLICATION." + vmFromDB.getDeployment().getApplication().getName() + ".DEPLOYMENT." + vmFromDB.getDeployment().getId() + ".VM." + vmFromDB.getId() + ".METRIC." + im.getName(); VM vm = new VM(); vm.setIaasVmId(vmID); vm.setMetricName(im.getName()); vm.setUnits(im.getUnits()); vm.setValue(im.getValue()); vm.setTimestamp(im.getTimestamp()); vm.setVmId("" + vmFromDB.getId()); vm.setProviderId(providerId); vm.setCpu(vmFromDB.getCpuActual()); vm.setRam(vmFromDB.getRamActual()); vm.setDisk(vmFromDB.getDiskActual()); vm.setSwap(vmFromDB.getSwapActual()); if(vmFromDB.getPriceSchema() != null) { vm.setPriceSchema(vmFromDB.getPriceSchema().intValue()); } ApplicationManagerMessage amM = new ApplicationManagerMessage(); amM.setApplicationId(vmFromDB.getDeployment().getApplication().getName()); amM.setDeploymentId("" + vmFromDB.getDeployment().getId()); amM.addVM(vm); logger.debug("Message to be sent: " + ModelConverter.applicationManagerMessageToJSON(amM)); AmqpProducer.sendMessage(topicToSend, amM, false); } } catch (JMSException e) { logger.info("Error receiving message from: " + host + " to the topic: " + topic); logger.info(e.getMessage()); } } MetricsListener(String amqpUrl, String topic, String providerId, VMDAO vmDAO); @Override void onMessage(Message message); void close(); }
MetricsListener implements MessageListener { @Override public void onMessage(Message message) { try { TextMessage textMessage = (TextMessage) message; String destination = textMessage.getJMSDestination().toString(); logger.debug("Message received for destionation: " + destination); String text = (String) textMessage.getText(); logger.debug("Text Message: " + text); IaaSMessage im = Converter.iaasMessageFromJSONToObject(text); logger.debug("DESTINATION: " + destination); String vmID = destination.split("\\.")[1]; logger.debug("VMDAO: " + vmDAO + " vmID " + vmID); eu.ascetic.paas.applicationmanager.model.VM vmFromDB = vmDAO.getVMWithProviderVMId(vmID, providerId); if(vmFromDB != null) { String topicToSend = "APPLICATION." + vmFromDB.getDeployment().getApplication().getName() + ".DEPLOYMENT." + vmFromDB.getDeployment().getId() + ".VM." + vmFromDB.getId() + ".METRIC." + im.getName(); VM vm = new VM(); vm.setIaasVmId(vmID); vm.setMetricName(im.getName()); vm.setUnits(im.getUnits()); vm.setValue(im.getValue()); vm.setTimestamp(im.getTimestamp()); vm.setVmId("" + vmFromDB.getId()); vm.setProviderId(providerId); vm.setCpu(vmFromDB.getCpuActual()); vm.setRam(vmFromDB.getRamActual()); vm.setDisk(vmFromDB.getDiskActual()); vm.setSwap(vmFromDB.getSwapActual()); if(vmFromDB.getPriceSchema() != null) { vm.setPriceSchema(vmFromDB.getPriceSchema().intValue()); } ApplicationManagerMessage amM = new ApplicationManagerMessage(); amM.setApplicationId(vmFromDB.getDeployment().getApplication().getName()); amM.setDeploymentId("" + vmFromDB.getDeployment().getId()); amM.addVM(vm); logger.debug("Message to be sent: " + ModelConverter.applicationManagerMessageToJSON(amM)); AmqpProducer.sendMessage(topicToSend, amM, false); } } catch (JMSException e) { logger.info("Error receiving message from: " + host + " to the topic: " + topic); logger.info(e.getMessage()); } } MetricsListener(String amqpUrl, String topic, String providerId, VMDAO vmDAO); @Override void onMessage(Message message); void close();  }