class 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(()->{
			// Merge the token and agent properties
			Map<String, String> mergedAgentProperties = getMergedAgentProperties(token);
			// Instantiate the function handler 
			String handlerClass = inputMessage.getProperties().get(FUNCTION_HANDLER_KEY);
			@SuppressWarnings("rawtypes")
			AbstractFunctionHandler functionHandler = functionHandlerFactory.create(applicationContextBuilder.getCurrentContext().getClassLoader(), 
					handlerClass, token.getSession(), token.getTokenReservationSession(), mergedAgentProperties);
			
			// Deserialize the Input from the message payload
			JavaType javaType = mapper.getTypeFactory().constructParametrizedType(Input.class, Input.class, functionHandler.getInputPayloadClass());
			Input<?> input = mapper.readValue(mapper.treeAsTokens(inputMessage.getPayload()), javaType);
			
			// Handle the input
			MeasurementsBuilder measurementsBuilder = new MeasurementsBuilder();
			measurementsBuilder.startMeasure(input.getFunction());
			@SuppressWarnings("unchecked")
			Output<?> output = functionHandler.handle(input);
			measurementsBuilder.stopMeasure(customMeasureData());
			
			List<Measure> outputMeasures = output.getMeasures();
			// Add type="custom" to all output measures
			addCustomTypeToOutputMeasures(outputMeasures);
			
			// Add Keyword measure to output
			addAdditionalMeasuresToOutput(output, measurementsBuilder.getMeasures());

			// Serialize the output
			ObjectNode outputPayload = (ObjectNode) mapper.valueToTree(output);

			// Create and return the output message 
			OutputMessageBuilder outputMessageBuilder = new OutputMessageBuilder();
			outputMessageBuilder.setPayload(outputPayload);
			return outputMessageBuilder.build();
			
		});
	}

	public  FunctionMessageHandler();

}