StellarAdapter implements EnrichmentAdapter<CacheKey>,Serializable { @Override public JSONObject enrich(CacheKey value) { Context stellarContext = (Context) value.getConfig().getConfiguration().get(STELLAR_CONTEXT_CONF); ConfigHandler handler = getHandler.apply(value.getConfig()); Map<String, Object> globalConfig = value.getConfig().getConfiguration(); Map<String, Object> sensorConfig = value.getConfig().getEnrichment().getConfig(); if(handler == null) { _LOG.trace("Stellar ConfigHandler is null."); return new JSONObject(); } Long slowLogThreshold = null; if(_PERF_LOG.isDebugEnabled()) { slowLogThreshold = ConversionUtils.convert(globalConfig.getOrDefault(STELLAR_SLOW_LOG, STELLAR_SLOW_LOG_DEFAULT), Long.class); } Map<String, Object> message = new HashMap<>(value.getValue(Map.class)); VariableResolver resolver = new MapVariableResolver(message, sensorConfig, globalConfig); StellarProcessor processor = new StellarProcessor(); JSONObject enriched = process(message , handler , value.getField() , slowLogThreshold , processor , resolver , stellarContext ); _LOG.trace("Stellar Enrichment Success: {}", enriched); return enriched; }  StellarAdapter ofType(String enrichmentType); @Override String getOutputPrefix(CacheKey value); @Override void logAccess(CacheKey value); @Override String getStreamSubGroup(String enrichmentType, String field); static Iterable<Map.Entry<String, Object>> getStellarStatements(ConfigHandler handler, String field); static JSONObject process( Map<String, Object> message
                                           , ConfigHandler handler
                                           , String field
                                           , Long slowLogThreshold
                                           , StellarProcessor processor
                                           , VariableResolver resolver
                                           , Context stellarContext
                                           ); @Override JSONObject enrich(CacheKey value); @Override boolean initializeAdapter(Map<String, Object> config); @Override void updateAdapter(Map<String, Object> config); @Override void cleanup(); static final String STELLAR_SLOW_LOG; static final Long STELLAR_SLOW_LOG_DEFAULT; }
@Test public void test_default() throws Exception { for(String c : DEFAULT_CONFIGS) { JSONObject message = getMessage(); EnrichmentConfig enrichmentConfig = JSONUtils.INSTANCE.load(c, EnrichmentConfig.class); assertNotNull(enrichmentConfig.getEnrichmentConfigs().get("stellar")); ConfigHandler handler = enrichmentConfig.getEnrichmentConfigs().get("stellar"); JSONObject enriched = enrich(message, "", handler); assertEquals("STELLAR_TEST", enriched.get("stmt1")); assertEquals("stellar_test", enriched.get("stmt2")); assertEquals("foo", enriched.get("stmt3")); assertEquals(3, enriched.size()); } }
@Test public void test_grouped() throws Exception { for(String c : GROUPED_CONFIGS) { JSONObject message = getMessage(); EnrichmentConfig enrichmentConfig = JSONUtils.INSTANCE.load(c, EnrichmentConfig.class); assertNotNull(enrichmentConfig.getEnrichmentConfigs().get("stellar")); ConfigHandler handler = enrichmentConfig.getEnrichmentConfigs().get("stellar"); { JSONObject enriched = enrich(message, "group1", handler); assertEquals("STELLAR_TEST", enriched.get("stmt1")); assertEquals("stellar_test", enriched.get("stmt2")); assertEquals(2, enriched.size()); } { JSONObject enriched = enrich(message, "group2", handler); assertEquals("foo", enriched.get("stmt3")); assertEquals(1, enriched.size()); } } }
@Test public void test_mixed() throws Exception { for(String c : MIXED_CONFIGS) { JSONObject message = getMessage(); EnrichmentConfig enrichmentConfig = JSONUtils.INSTANCE.load(c, EnrichmentConfig.class); assertNotNull(enrichmentConfig.getEnrichmentConfigs().get("stellar")); ConfigHandler handler = enrichmentConfig.getEnrichmentConfigs().get("stellar"); { JSONObject enriched = enrich(message, "group1", handler); assertEquals("STELLAR_TEST", enriched.get("stmt1")); assertEquals("stellar_test", enriched.get("stmt2")); assertEquals(2, enriched.size()); } { JSONObject enriched = enrich(message, "group2", handler); assertEquals("foo", enriched.get("stmt3")); assertEquals(1, enriched.size()); } { JSONObject enriched = enrich(message, "", handler); assertEquals(2, enriched.get("stmt4")); assertEquals("stellar_test", enriched.get("stmt5")); assertEquals(2, enriched.size()); } } }
@Test public void test_tempVariable() throws Exception { JSONObject message = getMessage(); EnrichmentConfig enrichmentConfig = JSONUtils.INSTANCE.load(tempVarStellarConfig_list, EnrichmentConfig.class); assertNotNull(enrichmentConfig.getEnrichmentConfigs().get("stellar")); ConfigHandler handler = enrichmentConfig.getEnrichmentConfigs().get("stellar"); { JSONObject enriched = enrich(message, "group1", handler); assertEquals("stellar_test", enriched.get("stmt2")); assertEquals(1, enriched.size()); } { JSONObject enriched = enrich(message, "group2", handler); assertEquals("foo", enriched.get("stmt3")); assertEquals(1, enriched.size()); } { JSONObject enriched = enrich(message, "", handler); assertEquals(2, enriched.get("stmt4")); assertEquals("stellar_test", enriched.get("stmt5")); assertEquals(2, enriched.size()); } }
@Test public void testAllVariableUsage() throws Exception { JSONObject message = getMessage(); EnrichmentConfig enrichmentConfig = JSONUtils.INSTANCE.load(allVariableConfig, EnrichmentConfig.class); assertNotNull(enrichmentConfig.getEnrichmentConfigs().get("stellar")); ConfigHandler handler = enrichmentConfig.getEnrichmentConfigs().get("stellar"); JSONObject enriched = enrich(message, "", handler); assertEquals("stellar_test", enriched.get("stmt1")); }