@Test public void testSubscribeToEntityWithAttributeWildcard() throws Exception { final CountDownLatch latch = new CountDownLatch(1); app.subscriptions().subscribe(entity, null, new SensorEventListener<Object>() { @Override public void onEvent(SensorEvent<Object> event) { latch.countDown(); }}); entity.setSequenceValue(1234); if (!latch.await(TIMEOUT_MS, TimeUnit.MILLISECONDS)) { fail("Timeout waiting for Event on TestEntity listener"); } }
@SuppressWarnings("unchecked") protected synchronized <T> SubscriptionHandle subscribe(Map<String, Object> flags, final Subscription<T> s) { Entity producer = s.producer; Sensor<T> sensor= s.sensor; s.subscriber = getSubscriber(flags, s); if (flags.containsKey("subscriberExecutionManagerTag")) { s.subscriberExecutionManagerTag = flags.remove("subscriberExecutionManagerTag"); s.subscriberExecutionManagerTagSupplied = true; } else { s.subscriberExecutionManagerTag = s.subscriber instanceof Entity ? "subscription-delivery-entity-"+((Entity)s.subscriber).getId()+"["+s.subscriber+"]" : s.subscriber instanceof String ? "subscription-delivery-string["+s.subscriber+"]" : "subscription-delivery-object["+s.subscriber+"]"; s.subscriberExecutionManagerTagSupplied = false; } s.eventFilter = (Predicate<SensorEvent<T>>) flags.remove("eventFilter"); boolean notifyOfInitialValue = Boolean.TRUE.equals(flags.remove("notifyOfInitialValue")); s.flags = flags; if (LOG.isDebugEnabled()) LOG.debug("Creating subscription {} for {} on {} {} in {}", new Object[] {s.id, s.subscriber, producer, sensor, this}); allSubscriptions.put(s.id, s); addToMapOfSets(subscriptionsByToken, makeEntitySensorToken(s.producer, s.sensor), s); if (s.subscriber!=null) { addToMapOfSets(subscriptionsBySubscriber, s.subscriber, s); } if (!s.subscriberExecutionManagerTagSupplied && s.subscriberExecutionManagerTag!=null) { ((BasicExecutionManager) em).setTaskSchedulerForTag(s.subscriberExecutionManagerTag, SingleThreadedScheduler.class); } if (notifyOfInitialValue) { if (producer == null) { LOG.warn("Cannot notifyOfInitialValue for subscription with wildcard producer: "+s); } else if (sensor == null) { LOG.warn("Cannot notifyOfInitialValue for subscription with wilcard sensor: "+s); } else if (!(sensor instanceof AttributeSensor)) { LOG.warn("Cannot notifyOfInitialValue for subscription with non-attribute sensor: "+s); } else { if (LOG.isTraceEnabled()) LOG.trace("sending initial value of {} -> {} to {}", new Object[] {s.producer, s.sensor, s}); Map<String, Object> tagsMap = MutableMap.of("tag", s.subscriberExecutionManagerTag); em.submit(tagsMap, new Runnable() { @Override public String toString() { return "LSM.publishInitialValue("+s.producer+", "+s.sensor+")"; } public void run() { Object val = s.producer.getAttribute((AttributeSensor<?>) s.sensor); @SuppressWarnings("rawtypes") SensorEvent event = new BasicSensorEvent(s.sensor, s.producer, val); if (s.eventFilter!=null && !s.eventFilter.apply(event)) return; try { s.listener.onEvent(event); } catch (Throwable t) { if (event!=null && event.getSource()!=null && Entities.isNoLongerManaged(event.getSource())) { LOG.debug("Error processing initial-value subscription to "+LocalSubscriptionManager.this+", after entity unmanaged: "+t, t); } else { LOG.warn("Error processing initial-value subscription to "+LocalSubscriptionManager.this+": "+t, t); } } }}); } } return s; }
LocalSubscriptionManager extends AbstractSubscriptionManager { @SuppressWarnings("unchecked") protected synchronized <T> SubscriptionHandle subscribe(Map<String, Object> flags, final Subscription<T> s) { Entity producer = s.producer; Sensor<T> sensor= s.sensor; s.subscriber = getSubscriber(flags, s); if (flags.containsKey("subscriberExecutionManagerTag")) { s.subscriberExecutionManagerTag = flags.remove("subscriberExecutionManagerTag"); s.subscriberExecutionManagerTagSupplied = true; } else { s.subscriberExecutionManagerTag = s.subscriber instanceof Entity ? "subscription-delivery-entity-"+((Entity)s.subscriber).getId()+"["+s.subscriber+"]" : s.subscriber instanceof String ? "subscription-delivery-string["+s.subscriber+"]" : "subscription-delivery-object["+s.subscriber+"]"; s.subscriberExecutionManagerTagSupplied = false; } s.eventFilter = (Predicate<SensorEvent<T>>) flags.remove("eventFilter"); boolean notifyOfInitialValue = Boolean.TRUE.equals(flags.remove("notifyOfInitialValue")); s.flags = flags; if (LOG.isDebugEnabled()) LOG.debug("Creating subscription {} for {} on {} {} in {}", new Object[] {s.id, s.subscriber, producer, sensor, this}); allSubscriptions.put(s.id, s); addToMapOfSets(subscriptionsByToken, makeEntitySensorToken(s.producer, s.sensor), s); if (s.subscriber!=null) { addToMapOfSets(subscriptionsBySubscriber, s.subscriber, s); } if (!s.subscriberExecutionManagerTagSupplied && s.subscriberExecutionManagerTag!=null) { ((BasicExecutionManager) em).setTaskSchedulerForTag(s.subscriberExecutionManagerTag, SingleThreadedScheduler.class); } if (notifyOfInitialValue) { if (producer == null) { LOG.warn("Cannot notifyOfInitialValue for subscription with wildcard producer: "+s); } else if (sensor == null) { LOG.warn("Cannot notifyOfInitialValue for subscription with wilcard sensor: "+s); } else if (!(sensor instanceof AttributeSensor)) { LOG.warn("Cannot notifyOfInitialValue for subscription with non-attribute sensor: "+s); } else { if (LOG.isTraceEnabled()) LOG.trace("sending initial value of {} -> {} to {}", new Object[] {s.producer, s.sensor, s}); Map<String, Object> tagsMap = MutableMap.of("tag", s.subscriberExecutionManagerTag); em.submit(tagsMap, new Runnable() { @Override public String toString() { return "LSM.publishInitialValue("+s.producer+", "+s.sensor+")"; } public void run() { Object val = s.producer.getAttribute((AttributeSensor<?>) s.sensor); @SuppressWarnings("rawtypes") SensorEvent event = new BasicSensorEvent(s.sensor, s.producer, val); if (s.eventFilter!=null && !s.eventFilter.apply(event)) return; try { s.listener.onEvent(event); } catch (Throwable t) { if (event!=null && event.getSource()!=null && Entities.isNoLongerManaged(event.getSource())) { LOG.debug("Error processing initial-value subscription to "+LocalSubscriptionManager.this+", after entity unmanaged: "+t, t); } else { LOG.warn("Error processing initial-value subscription to "+LocalSubscriptionManager.this+": "+t, t); } } }}); } } return s; } }
LocalSubscriptionManager extends AbstractSubscriptionManager { @SuppressWarnings("unchecked") protected synchronized <T> SubscriptionHandle subscribe(Map<String, Object> flags, final Subscription<T> s) { Entity producer = s.producer; Sensor<T> sensor= s.sensor; s.subscriber = getSubscriber(flags, s); if (flags.containsKey("subscriberExecutionManagerTag")) { s.subscriberExecutionManagerTag = flags.remove("subscriberExecutionManagerTag"); s.subscriberExecutionManagerTagSupplied = true; } else { s.subscriberExecutionManagerTag = s.subscriber instanceof Entity ? "subscription-delivery-entity-"+((Entity)s.subscriber).getId()+"["+s.subscriber+"]" : s.subscriber instanceof String ? "subscription-delivery-string["+s.subscriber+"]" : "subscription-delivery-object["+s.subscriber+"]"; s.subscriberExecutionManagerTagSupplied = false; } s.eventFilter = (Predicate<SensorEvent<T>>) flags.remove("eventFilter"); boolean notifyOfInitialValue = Boolean.TRUE.equals(flags.remove("notifyOfInitialValue")); s.flags = flags; if (LOG.isDebugEnabled()) LOG.debug("Creating subscription {} for {} on {} {} in {}", new Object[] {s.id, s.subscriber, producer, sensor, this}); allSubscriptions.put(s.id, s); addToMapOfSets(subscriptionsByToken, makeEntitySensorToken(s.producer, s.sensor), s); if (s.subscriber!=null) { addToMapOfSets(subscriptionsBySubscriber, s.subscriber, s); } if (!s.subscriberExecutionManagerTagSupplied && s.subscriberExecutionManagerTag!=null) { ((BasicExecutionManager) em).setTaskSchedulerForTag(s.subscriberExecutionManagerTag, SingleThreadedScheduler.class); } if (notifyOfInitialValue) { if (producer == null) { LOG.warn("Cannot notifyOfInitialValue for subscription with wildcard producer: "+s); } else if (sensor == null) { LOG.warn("Cannot notifyOfInitialValue for subscription with wilcard sensor: "+s); } else if (!(sensor instanceof AttributeSensor)) { LOG.warn("Cannot notifyOfInitialValue for subscription with non-attribute sensor: "+s); } else { if (LOG.isTraceEnabled()) LOG.trace("sending initial value of {} -> {} to {}", new Object[] {s.producer, s.sensor, s}); Map<String, Object> tagsMap = MutableMap.of("tag", s.subscriberExecutionManagerTag); em.submit(tagsMap, new Runnable() { @Override public String toString() { return "LSM.publishInitialValue("+s.producer+", "+s.sensor+")"; } public void run() { Object val = s.producer.getAttribute((AttributeSensor<?>) s.sensor); @SuppressWarnings("rawtypes") SensorEvent event = new BasicSensorEvent(s.sensor, s.producer, val); if (s.eventFilter!=null && !s.eventFilter.apply(event)) return; try { s.listener.onEvent(event); } catch (Throwable t) { if (event!=null && event.getSource()!=null && Entities.isNoLongerManaged(event.getSource())) { LOG.debug("Error processing initial-value subscription to "+LocalSubscriptionManager.this+", after entity unmanaged: "+t, t); } else { LOG.warn("Error processing initial-value subscription to "+LocalSubscriptionManager.this+": "+t, t); } } }}); } } return s; } LocalSubscriptionManager(ExecutionManager m); }
LocalSubscriptionManager extends AbstractSubscriptionManager { @SuppressWarnings("unchecked") protected synchronized <T> SubscriptionHandle subscribe(Map<String, Object> flags, final Subscription<T> s) { Entity producer = s.producer; Sensor<T> sensor= s.sensor; s.subscriber = getSubscriber(flags, s); if (flags.containsKey("subscriberExecutionManagerTag")) { s.subscriberExecutionManagerTag = flags.remove("subscriberExecutionManagerTag"); s.subscriberExecutionManagerTagSupplied = true; } else { s.subscriberExecutionManagerTag = s.subscriber instanceof Entity ? "subscription-delivery-entity-"+((Entity)s.subscriber).getId()+"["+s.subscriber+"]" : s.subscriber instanceof String ? "subscription-delivery-string["+s.subscriber+"]" : "subscription-delivery-object["+s.subscriber+"]"; s.subscriberExecutionManagerTagSupplied = false; } s.eventFilter = (Predicate<SensorEvent<T>>) flags.remove("eventFilter"); boolean notifyOfInitialValue = Boolean.TRUE.equals(flags.remove("notifyOfInitialValue")); s.flags = flags; if (LOG.isDebugEnabled()) LOG.debug("Creating subscription {} for {} on {} {} in {}", new Object[] {s.id, s.subscriber, producer, sensor, this}); allSubscriptions.put(s.id, s); addToMapOfSets(subscriptionsByToken, makeEntitySensorToken(s.producer, s.sensor), s); if (s.subscriber!=null) { addToMapOfSets(subscriptionsBySubscriber, s.subscriber, s); } if (!s.subscriberExecutionManagerTagSupplied && s.subscriberExecutionManagerTag!=null) { ((BasicExecutionManager) em).setTaskSchedulerForTag(s.subscriberExecutionManagerTag, SingleThreadedScheduler.class); } if (notifyOfInitialValue) { if (producer == null) { LOG.warn("Cannot notifyOfInitialValue for subscription with wildcard producer: "+s); } else if (sensor == null) { LOG.warn("Cannot notifyOfInitialValue for subscription with wilcard sensor: "+s); } else if (!(sensor instanceof AttributeSensor)) { LOG.warn("Cannot notifyOfInitialValue for subscription with non-attribute sensor: "+s); } else { if (LOG.isTraceEnabled()) LOG.trace("sending initial value of {} -> {} to {}", new Object[] {s.producer, s.sensor, s}); Map<String, Object> tagsMap = MutableMap.of("tag", s.subscriberExecutionManagerTag); em.submit(tagsMap, new Runnable() { @Override public String toString() { return "LSM.publishInitialValue("+s.producer+", "+s.sensor+")"; } public void run() { Object val = s.producer.getAttribute((AttributeSensor<?>) s.sensor); @SuppressWarnings("rawtypes") SensorEvent event = new BasicSensorEvent(s.sensor, s.producer, val); if (s.eventFilter!=null && !s.eventFilter.apply(event)) return; try { s.listener.onEvent(event); } catch (Throwable t) { if (event!=null && event.getSource()!=null && Entities.isNoLongerManaged(event.getSource())) { LOG.debug("Error processing initial-value subscription to "+LocalSubscriptionManager.this+", after entity unmanaged: "+t, t); } else { LOG.warn("Error processing initial-value subscription to "+LocalSubscriptionManager.this+": "+t, t); } } }}); } } return s; } LocalSubscriptionManager(ExecutionManager m); long getNumSubscriptions(); long getTotalEventsPublished(); long getTotalEventsDelivered(); @SuppressWarnings("unchecked") Set<SubscriptionHandle> getSubscriptionsForSubscriber(Object subscriber); synchronized Set<SubscriptionHandle> getSubscriptionsForEntitySensor(Entity source, Sensor<?> sensor); @SuppressWarnings("rawtypes") synchronized boolean unsubscribe(SubscriptionHandle sh); @SuppressWarnings({ "unchecked", "rawtypes" }) void publish(final SensorEvent<T> event); @Override String toString(); }
LocalSubscriptionManager extends AbstractSubscriptionManager { @SuppressWarnings("unchecked") protected synchronized <T> SubscriptionHandle subscribe(Map<String, Object> flags, final Subscription<T> s) { Entity producer = s.producer; Sensor<T> sensor= s.sensor; s.subscriber = getSubscriber(flags, s); if (flags.containsKey("subscriberExecutionManagerTag")) { s.subscriberExecutionManagerTag = flags.remove("subscriberExecutionManagerTag"); s.subscriberExecutionManagerTagSupplied = true; } else { s.subscriberExecutionManagerTag = s.subscriber instanceof Entity ? "subscription-delivery-entity-"+((Entity)s.subscriber).getId()+"["+s.subscriber+"]" : s.subscriber instanceof String ? "subscription-delivery-string["+s.subscriber+"]" : "subscription-delivery-object["+s.subscriber+"]"; s.subscriberExecutionManagerTagSupplied = false; } s.eventFilter = (Predicate<SensorEvent<T>>) flags.remove("eventFilter"); boolean notifyOfInitialValue = Boolean.TRUE.equals(flags.remove("notifyOfInitialValue")); s.flags = flags; if (LOG.isDebugEnabled()) LOG.debug("Creating subscription {} for {} on {} {} in {}", new Object[] {s.id, s.subscriber, producer, sensor, this}); allSubscriptions.put(s.id, s); addToMapOfSets(subscriptionsByToken, makeEntitySensorToken(s.producer, s.sensor), s); if (s.subscriber!=null) { addToMapOfSets(subscriptionsBySubscriber, s.subscriber, s); } if (!s.subscriberExecutionManagerTagSupplied && s.subscriberExecutionManagerTag!=null) { ((BasicExecutionManager) em).setTaskSchedulerForTag(s.subscriberExecutionManagerTag, SingleThreadedScheduler.class); } if (notifyOfInitialValue) { if (producer == null) { LOG.warn("Cannot notifyOfInitialValue for subscription with wildcard producer: "+s); } else if (sensor == null) { LOG.warn("Cannot notifyOfInitialValue for subscription with wilcard sensor: "+s); } else if (!(sensor instanceof AttributeSensor)) { LOG.warn("Cannot notifyOfInitialValue for subscription with non-attribute sensor: "+s); } else { if (LOG.isTraceEnabled()) LOG.trace("sending initial value of {} -> {} to {}", new Object[] {s.producer, s.sensor, s}); Map<String, Object> tagsMap = MutableMap.of("tag", s.subscriberExecutionManagerTag); em.submit(tagsMap, new Runnable() { @Override public String toString() { return "LSM.publishInitialValue("+s.producer+", "+s.sensor+")"; } public void run() { Object val = s.producer.getAttribute((AttributeSensor<?>) s.sensor); @SuppressWarnings("rawtypes") SensorEvent event = new BasicSensorEvent(s.sensor, s.producer, val); if (s.eventFilter!=null && !s.eventFilter.apply(event)) return; try { s.listener.onEvent(event); } catch (Throwable t) { if (event!=null && event.getSource()!=null && Entities.isNoLongerManaged(event.getSource())) { LOG.debug("Error processing initial-value subscription to "+LocalSubscriptionManager.this+", after entity unmanaged: "+t, t); } else { LOG.warn("Error processing initial-value subscription to "+LocalSubscriptionManager.this+": "+t, t); } } }}); } } return s; } LocalSubscriptionManager(ExecutionManager m); long getNumSubscriptions(); long getTotalEventsPublished(); long getTotalEventsDelivered(); @SuppressWarnings("unchecked") Set<SubscriptionHandle> getSubscriptionsForSubscriber(Object subscriber); synchronized Set<SubscriptionHandle> getSubscriptionsForEntitySensor(Entity source, Sensor<?> sensor); @SuppressWarnings("rawtypes") synchronized boolean unsubscribe(SubscriptionHandle sh); @SuppressWarnings({ "unchecked", "rawtypes" }) void publish(final SensorEvent<T> event); @Override String toString();  }