EvalFunctionWrapper extends ProcessFunction<MRecord, GenericEvent> implements GenericEventHandler, Serializable { Map<String, Value> createTypedChannelMap(MRecord value) { List<String> notFoundKeys = channelsAndTypes.entrySet().stream() .filter(entry -> !value.getChannels().contains(entry.getKey())) .map(Map.Entry::getKey) .collect(Collectors.toList()); if (!notFoundKeys.isEmpty()) { logger.trace("No key matching SPS parameter found for evaluation {}, the missing channels are {}", call.getEvalFunction().getFunctionDef().getSignature().getName(), notFoundKeys); return Collections.emptyMap(); } return channelsAndTypes.entrySet().stream() .collect(Collectors.toMap(Map.Entry::getKey, entry -> value.getValue(entry.getKey()))); } EvalFunctionWrapper(EvalFunctionCall call); @Override void open(Configuration parameters); @Override void processElement(MRecord value, Context ctx, Collector<GenericEvent> out); @Override void fire(GenericEvent event);  }
@Test public void createTypedChannelMap_stringToString_works() { EvalFunctionWrapper evalFunctionWrapper = new EvalFunctionWrapper(call); Map<String, Value> typedChannelMap = evalFunctionWrapper.createTypedChannelMap(typedValues); assertEquals(1, typedChannelMap.size()); assertEquals("a_string", typedChannelMap.get("DB13_StringChannel").getAsString()); }
@Test public void createTypedChannelMap_longToString_works() throws Exception { EvalFunctionCall call = new EvalFunctionCall( new AnnotatedEvalFunctionWrapper<>(RegexFind2.class), Collections.singletonMap("regex", Value.of("nothing")), Collections.singletonMap("value", "DB13_LongChannel")); EvalFunctionWrapper evalFunctionWrapper = new EvalFunctionWrapper(call); Map<String, Value> typedChannelMap = evalFunctionWrapper.createTypedChannelMap(typedValues); assertEquals(1, typedChannelMap.size()); assertEquals("100", typedChannelMap.get("DB13_LongChannel").getAsString()); }