@Test public void postMessageOnEventUpdate() { HttpHeaders httpHeaders = new HttpHeaders(); httpHeaders.setContentType(MediaType.APPLICATION_JSON); httpHeaders.setAccept(Collections.singletonList(MediaType.APPLICATION_JSON)); when(statement.getText()).thenReturn("statement"); when(ngsiClient.getRequestHeaders(any())).thenReturn(httpHeaders); List<ContextAttribute> attributes = new LinkedList<>(); attributes.add(new ContextAttribute("id", "string", "OUT1234")); attributes.add(new ContextAttribute("avgTemp", "double", 10.25)); attributes.add(new ContextAttribute("avgTemp_unit","string","celcius")); EventBean[]beans = {buildEventBean("TempSensorAvg", attributes)}; eventSinkListener.update(beans, null, statement, provider); ArgumentCaptor<UpdateContext> updateContextArg = ArgumentCaptor.forClass(UpdateContext.class); ArgumentCaptor<HttpHeaders> headersArg = ArgumentCaptor.forClass(HttpHeaders.class); verify(ngsiClient).updateContext(eq(broker.getUrl()), headersArg.capture(), updateContextArg.capture()); UpdateContext updateContext = updateContextArg.getValue(); assertEquals(UpdateAction.APPEND, updateContext.getUpdateAction()); assertEquals(1, updateContext.getContextElements().size()); HttpHeaders headers = headersArg.getValue(); assertEquals(MediaType.APPLICATION_JSON, headers.getContentType()); assertTrue(headers.getAccept().contains(MediaType.APPLICATION_JSON)); assertEquals("SN", headers.getFirst("Fiware-Service")); assertEquals("SP", headers.getFirst("Fiware-ServicePath")); assertEquals("AUTH_TOKEN", headers.getFirst("X-Auth-Token")); ContextElement contextElement = updateContext.getContextElements().get(0); assertEquals("OUT1234", contextElement.getEntityId().getId()); assertEquals("TempSensorAvg", contextElement.getEntityId().getType()); assertFalse(contextElement.getEntityId().getIsPattern()); assertEquals(1, contextElement.getContextAttributeList().size()); ContextAttribute attr = contextElement.getContextAttributeList().get(0); assertEquals("avgTemp", attr.getName()); assertEquals("double", attr.getType()); assertEquals(10.25, attr.getValue()); assertEquals(1, attr.getMetadata().size()); assertEquals("unit", attr.getMetadata().get(0).getName()); assertEquals("string", attr.getMetadata().get(0).getType()); assertEquals("celcius", attr.getMetadata().get(0).getValue()); }
@Override public void update(EventBean[] added, EventBean[] removed, EPStatement epStatement, EPServiceProvider epServiceProvider) { if (added == null) { return; } for (EventBean eventBean : added) { if (logger.isInfoEnabled()) { StringBuilder sb = new StringBuilder(); for (String propertyName : eventBean.getEventType().getPropertyNames()) { sb.append(" / ").append(propertyName).append(':').append(eventBean.get(propertyName)); } logger.info("EventOut: {}{} from {}", eventBean.getEventType().getName(), sb.toString(), epStatement.getText()); } final String type = eventBean.getEventType().getName(); final EventTypeOut eventTypeOut = getEventTypeOut(type); if (eventTypeOut == null) { logger.debug("EventTypeOut {} doesn't exist in Configuration, updateContext not sent", type); } else { UpdateContext updateContext = buildUpdateContextRequest(eventBean, eventTypeOut); if (updateContext != null) { for (Broker broker : eventTypeOut.getBrokers()) { assert broker != null; HttpHeaders httpHeaders = getHeadersForBroker(broker); ngsiClient.updateContext(broker.getUrl(), httpHeaders, updateContext).addCallback( updateContextResponse -> logger.debug("UpdateContext completed for {}", broker.getUrl()), throwable -> logger.warn("UpdateContext failed for {}", broker.getUrl(), throwable) ); } } } } }
EventSinkListener implements StatementAwareUpdateListener { @Override public void update(EventBean[] added, EventBean[] removed, EPStatement epStatement, EPServiceProvider epServiceProvider) { if (added == null) { return; } for (EventBean eventBean : added) { if (logger.isInfoEnabled()) { StringBuilder sb = new StringBuilder(); for (String propertyName : eventBean.getEventType().getPropertyNames()) { sb.append(" / ").append(propertyName).append(':').append(eventBean.get(propertyName)); } logger.info("EventOut: {}{} from {}", eventBean.getEventType().getName(), sb.toString(), epStatement.getText()); } final String type = eventBean.getEventType().getName(); final EventTypeOut eventTypeOut = getEventTypeOut(type); if (eventTypeOut == null) { logger.debug("EventTypeOut {} doesn't exist in Configuration, updateContext not sent", type); } else { UpdateContext updateContext = buildUpdateContextRequest(eventBean, eventTypeOut); if (updateContext != null) { for (Broker broker : eventTypeOut.getBrokers()) { assert broker != null; HttpHeaders httpHeaders = getHeadersForBroker(broker); ngsiClient.updateContext(broker.getUrl(), httpHeaders, updateContext).addCallback( updateContextResponse -> logger.debug("UpdateContext completed for {}", broker.getUrl()), throwable -> logger.warn("UpdateContext failed for {}", broker.getUrl(), throwable) ); } } } } } }
EventSinkListener implements StatementAwareUpdateListener { @Override public void update(EventBean[] added, EventBean[] removed, EPStatement epStatement, EPServiceProvider epServiceProvider) { if (added == null) { return; } for (EventBean eventBean : added) { if (logger.isInfoEnabled()) { StringBuilder sb = new StringBuilder(); for (String propertyName : eventBean.getEventType().getPropertyNames()) { sb.append(" / ").append(propertyName).append(':').append(eventBean.get(propertyName)); } logger.info("EventOut: {}{} from {}", eventBean.getEventType().getName(), sb.toString(), epStatement.getText()); } final String type = eventBean.getEventType().getName(); final EventTypeOut eventTypeOut = getEventTypeOut(type); if (eventTypeOut == null) { logger.debug("EventTypeOut {} doesn't exist in Configuration, updateContext not sent", type); } else { UpdateContext updateContext = buildUpdateContextRequest(eventBean, eventTypeOut); if (updateContext != null) { for (Broker broker : eventTypeOut.getBrokers()) { assert broker != null; HttpHeaders httpHeaders = getHeadersForBroker(broker); ngsiClient.updateContext(broker.getUrl(), httpHeaders, updateContext).addCallback( updateContextResponse -> logger.debug("UpdateContext completed for {}", broker.getUrl()), throwable -> logger.warn("UpdateContext failed for {}", broker.getUrl(), throwable) ); } } } } }  }
EventSinkListener implements StatementAwareUpdateListener { @Override public void update(EventBean[] added, EventBean[] removed, EPStatement epStatement, EPServiceProvider epServiceProvider) { if (added == null) { return; } for (EventBean eventBean : added) { if (logger.isInfoEnabled()) { StringBuilder sb = new StringBuilder(); for (String propertyName : eventBean.getEventType().getPropertyNames()) { sb.append(" / ").append(propertyName).append(':').append(eventBean.get(propertyName)); } logger.info("EventOut: {}{} from {}", eventBean.getEventType().getName(), sb.toString(), epStatement.getText()); } final String type = eventBean.getEventType().getName(); final EventTypeOut eventTypeOut = getEventTypeOut(type); if (eventTypeOut == null) { logger.debug("EventTypeOut {} doesn't exist in Configuration, updateContext not sent", type); } else { UpdateContext updateContext = buildUpdateContextRequest(eventBean, eventTypeOut); if (updateContext != null) { for (Broker broker : eventTypeOut.getBrokers()) { assert broker != null; HttpHeaders httpHeaders = getHeadersForBroker(broker); ngsiClient.updateContext(broker.getUrl(), httpHeaders, updateContext).addCallback( updateContextResponse -> logger.debug("UpdateContext completed for {}", broker.getUrl()), throwable -> logger.warn("UpdateContext failed for {}", broker.getUrl(), throwable) ); } } } } }  @Override void update(EventBean[] added, EventBean[] removed, EPStatement epStatement, EPServiceProvider epServiceProvider); void setConfiguration(Configuration configuration); }
EventSinkListener implements StatementAwareUpdateListener { @Override public void update(EventBean[] added, EventBean[] removed, EPStatement epStatement, EPServiceProvider epServiceProvider) { if (added == null) { return; } for (EventBean eventBean : added) { if (logger.isInfoEnabled()) { StringBuilder sb = new StringBuilder(); for (String propertyName : eventBean.getEventType().getPropertyNames()) { sb.append(" / ").append(propertyName).append(':').append(eventBean.get(propertyName)); } logger.info("EventOut: {}{} from {}", eventBean.getEventType().getName(), sb.toString(), epStatement.getText()); } final String type = eventBean.getEventType().getName(); final EventTypeOut eventTypeOut = getEventTypeOut(type); if (eventTypeOut == null) { logger.debug("EventTypeOut {} doesn't exist in Configuration, updateContext not sent", type); } else { UpdateContext updateContext = buildUpdateContextRequest(eventBean, eventTypeOut); if (updateContext != null) { for (Broker broker : eventTypeOut.getBrokers()) { assert broker != null; HttpHeaders httpHeaders = getHeadersForBroker(broker); ngsiClient.updateContext(broker.getUrl(), httpHeaders, updateContext).addCallback( updateContextResponse -> logger.debug("UpdateContext completed for {}", broker.getUrl()), throwable -> logger.warn("UpdateContext failed for {}", broker.getUrl(), throwable) ); } } } } }  @Override void update(EventBean[] added, EventBean[] removed, EPStatement epStatement, EPServiceProvider epServiceProvider); void setConfiguration(Configuration configuration);  }