MQTTNotificationStrategy implements NotificationStrategy { @Override public void execute(NotificationContext ctx) throws PushNotificationExecutionFailedException { String adapterName = mqttAdapterName; String tenantDomain = PrivilegedCarbonContext.getThreadLocalCarbonContext().getTenantDomain(true); if (!providerTenantDomain.equals(tenantDomain)) { adapterName = "mqtt.adapter." + PrivilegedCarbonContext.getThreadLocalCarbonContext().getTenantDomain() .toLowerCase(); try { MQTTDataHolder.getInstance().getOutputEventAdapterService().isPolled(adapterName); } catch (OutputEventAdapterException e) { synchronized (lockObj) { OutputEventAdapterConfiguration adapterConfig = new OutputEventAdapterConfiguration(); adapterConfig.setType(MQTTAdapterConstants.MQTT_ADAPTER_TYPE); adapterConfig.setMessageFormat(MessageType.TEXT); adapterConfig.setName(adapterName); Map<String, String> configProperties = new HashMap<String, String>(); adapterConfig.setStaticProperties(configProperties); try { MQTTDataHolder.getInstance().getOutputEventAdapterService().create(adapterConfig); } catch (OutputEventAdapterException e1) { throw new PushNotificationExecutionFailedException ("Error occurred while initializing MQTT output event adapter for shared tenant: " + tenantDomain, e); } } } } Operation operation = ctx.getOperation(); Properties properties = operation.getProperties(); if (properties != null && properties.get(MQTT_ADAPTER_TOPIC) != null) { Map<String, String> dynamicProperties = new HashMap<>(); dynamicProperties.put("topic", (String) properties.get(MQTT_ADAPTER_TOPIC)); MQTTDataHolder.getInstance().getOutputEventAdapterService().publish(adapterName, dynamicProperties, operation.getPayLoad()); } else { if (PolicyOperation.POLICY_OPERATION_CODE.equals(operation.getCode())) { PolicyOperation policyOperation = (PolicyOperation) operation; List<ProfileOperation> profileOperations = policyOperation.getProfileOperations(); String deviceType = ctx.getDeviceId().getType(); String deviceId = ctx.getDeviceId().getId(); for (ProfileOperation profileOperation : profileOperations) { Map<String, String> dynamicProperties = new HashMap<>(); String topic = tenantDomain + "/" + deviceType + "/" + deviceId + "/operation/" + profileOperation.getType() .toString().toLowerCase() + "/" + profileOperation.getCode().toLowerCase(); dynamicProperties.put("topic", topic); MQTTDataHolder.getInstance().getOutputEventAdapterService().publish(adapterName, dynamicProperties, profileOperation.getPayLoad()); } } else { Map<String, String> dynamicProperties = new HashMap<>(); String topic = PrivilegedCarbonContext.getThreadLocalCarbonContext().getTenantDomain(true) + "/" + ctx.getDeviceId().getType() + "/" + ctx.getDeviceId().getId() + "/operation/" + operation.getType().toString().toLowerCase() + "/" + operation.getCode() + "/" + operation.getId(); dynamicProperties.put("topic", topic); Object payload; if ("command".equals(operation.getType().toString().toLowerCase())) { payload = operation.getCode(); } else { payload = operation.getPayLoad(); } MQTTDataHolder.getInstance().getOutputEventAdapterService().publish(adapterName, dynamicProperties, payload); } } } MQTTNotificationStrategy(PushNotificationConfig config); @Override void init(); @Override void execute(NotificationContext ctx); @Override NotificationContext buildContext(); @Override void undeploy(); @Override PushNotificationConfig getConfig();  }
@Test(description = "testing execute method without properties and operation type command") public void testExecuteWithoutProperties() throws PushNotificationExecutionFailedException { deviceIdentifier = new DeviceIdentifier(); deviceIdentifier.setId("1"); deviceIdentifier.setType("SampleDeviceType"); operation = new Operation(); operation.setPayLoad(new Object()); operation.setType(Operation.Type.COMMAND); operation.setCode("SampleCode"); operation.setId(1); propertiesMap = new HashMap<>(); propertiesMap.put(MQTT_ADAPTER_TOPIC, "SampleTopic"); notificationContext = new NotificationContext(deviceIdentifier, operation); notificationContext.setProperties(propertiesMap); mqttNotificationStrategy.execute(notificationContext); }
@Test(dependsOnMethods = "testExecuteWithoutProperties", description = "testing execute method without properties " + "and operation type config") public void testExecutionWithoutPropertiesNonCommandType() throws PushNotificationExecutionFailedException { operation.setType(Operation.Type.CONFIG); operation.setProperties(null); notificationContext = new NotificationContext(deviceIdentifier, operation); mqttNotificationStrategy.execute(notificationContext); }
@Test(dependsOnMethods = {"testExecutionWithoutPropertiesNonCommandType"}, description = "test execute method " + "with a profile operation") public void testExecutePolicyOperation() throws PushNotificationExecutionFailedException { PolicyOperation policyOperation = new PolicyOperation(); policyOperation.setCode(POLICY_OPERATION_CODE); policyOperation.setProperties(null); ProfileOperation profileOperation = new ProfileOperation(); profileOperation.setActivityId("1"); profileOperation.setCode("SampleCode"); List<ProfileOperation> profileOperationList = new ArrayList<>(); profileOperationList.add(profileOperation); policyOperation.setProfileOperations(profileOperationList); notificationContext = new NotificationContext(deviceIdentifier, policyOperation); mqttNotificationStrategy.execute(notificationContext); }
@Test(dependsOnMethods = "testExecuteWithoutProperties", description = "testing execute method with properties") public void testExecute() throws PushNotificationExecutionFailedException { Properties properties = new Properties(); properties.setProperty(MQTT_ADAPTER_TOPIC, "SampleTopic"); operation.setProperties(properties); notificationContext = new NotificationContext(deviceIdentifier, operation); notificationContext.setProperties(propertiesMap); mqttNotificationStrategy.execute(notificationContext); }
@Test(dependsOnMethods = {"testExecute"}, description = "testing execute method without the default tenant domain") public void testExecutionWithoutTenantDomain() throws NoSuchFieldException, IllegalAccessException, PushNotificationExecutionFailedException { Field providerTenantDomain = MQTTNotificationStrategy.class.getDeclaredField("providerTenantDomain"); providerTenantDomain.setAccessible(true); providerTenantDomain.set(mqttNotificationStrategy, "SampleTenantDomain"); mqttNotificationStrategy.execute(notificationContext); }