@Test public void fallbackOnConfigurationId() { when(statement.getText()).thenReturn("statement"); when(ngsiClient.getRequestHeaders()).thenReturn(new HttpHeaders()); List<ContextAttribute> attributes = new LinkedList<>(); attributes.add(new ContextAttribute("id", "string", null)); attributes.add(new ContextAttribute("avgTemp", "double", 10.25)); EventBean[]beans = {buildEventBean("TempSensorAvg", attributes)}; eventSinkListener.update(beans, null, statement, provider); ArgumentCaptor<UpdateContext> updateContextArg = ArgumentCaptor.forClass(UpdateContext.class); verify(ngsiClient).updateContext(eq(broker.getUrl()), any(), updateContextArg.capture()); ContextElement contextElement = updateContextArg.getValue().getContextElements().get(0); assertEquals("OUT1", contextElement.getEntityId().getId()); }
@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);  }