FunctionMessageHandler extends AbstractMessageHandler { @Override public OutputMessage handle(AgentTokenWrapper token, InputMessage inputMessage) throws Exception { applicationContextBuilder.resetContext(); FileVersionId functionPackage = getFileVersionId(FUNCTION_HANDLER_PACKAGE_KEY, inputMessage.getProperties()); if(functionPackage != null) { RemoteApplicationContextFactory functionHandlerContext = new RemoteApplicationContextFactory(token.getServices().getFileManagerClient(), getFileVersionId(FUNCTION_HANDLER_PACKAGE_KEY, inputMessage.getProperties())); applicationContextBuilder.pushContext(functionHandlerContext); } return applicationContextBuilder.runInContext(()->{ Map<String, String> mergedAgentProperties = getMergedAgentProperties(token); String handlerClass = inputMessage.getProperties().get(FUNCTION_HANDLER_KEY); @SuppressWarnings("rawtypes") AbstractFunctionHandler functionHandler = functionHandlerFactory.create(applicationContextBuilder.getCurrentContext().getClassLoader(), handlerClass, token.getSession(), token.getTokenReservationSession(), mergedAgentProperties); JavaType javaType = mapper.getTypeFactory().constructParametrizedType(Input.class, Input.class, functionHandler.getInputPayloadClass()); Input<?> input = mapper.readValue(mapper.treeAsTokens(inputMessage.getPayload()), javaType); MeasurementsBuilder measurementsBuilder = new MeasurementsBuilder(); measurementsBuilder.startMeasure(input.getFunction()); @SuppressWarnings("unchecked") Output<?> output = functionHandler.handle(input); measurementsBuilder.stopMeasure(customMeasureData()); List<Measure> outputMeasures = output.getMeasures(); addCustomTypeToOutputMeasures(outputMeasures); addAdditionalMeasuresToOutput(output, measurementsBuilder.getMeasures()); ObjectNode outputPayload = (ObjectNode) mapper.valueToTree(output); OutputMessageBuilder outputMessageBuilder = new OutputMessageBuilder(); outputMessageBuilder.setPayload(outputPayload); return outputMessageBuilder.build(); }); } FunctionMessageHandler(); @Override void init(AgentTokenServices agentTokenServices); @Override OutputMessage handle(AgentTokenWrapper token, InputMessage inputMessage); static final String FUNCTION_HANDLER_PACKAGE_KEY; static final String FUNCTION_HANDLER_KEY; public FunctionHandlerFactory functionHandlerFactory; }
@Test public void test() throws Exception { AgentTokenWrapper agentToken = getAgentToken(tokenServices); InputMessage message = new InputMessage(); HashMap<String, String> properties = new HashMap<String, String>(); properties.put(FunctionMessageHandler.FUNCTION_HANDLER_PACKAGE_KEY + ".id", EMPTY_FILE); properties.put(FunctionMessageHandler.FUNCTION_HANDLER_PACKAGE_KEY + ".version", "1"); properties.put(FunctionMessageHandler.FUNCTION_HANDLER_KEY, TestFunctionHandler.class.getName()); message.setProperties(properties); Input<TestInput> input = getTestInput(); message.setPayload(new ObjectMapper().valueToTree(input)); OutputMessage outputMessage = messageHandlerPool.get(FunctionMessageHandler.class.getName()).handle(agentToken, message); Assert.assertEquals("Bonjour", outputMessage.getPayload().get("payload").get("message").asText()); }