LocalSubscriptionManager extends AbstractSubscriptionManager { @Override @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); s.subscriptionDescription = getSubscriptionDescription(flags, s); if (flags.containsKey("tags") || flags.containsKey("tag")) { Iterable<?> tags = (Iterable<?>) flags.get("tags"); Object tag = flags.get("tag"); s.subscriberExtraExecTags = (tag == null) ? tags : (tags == null ? ImmutableList.of(tag) : MutableList.builder().addAll(tags).add(tag).build()); } 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 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); T lastVal; if (notifyOfInitialValue) { notifyOfInitialValue = false; 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 { notifyOfInitialValue = true; } } if (notifyOfInitialValue) { lastVal = (T) s.producer.sensors().get((AttributeSensor<?>) s.sensor); } else { lastVal = null; } 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 (LOG.isTraceEnabled()) LOG.trace("sending initial value of {} -> {} to {}", new Object[] {s.producer, s.sensor, s}); em.submit( MutableMap.of("tags", getPublishTags(s, s.producer), "displayName", "Initial value publication on subscription to "+s.sensor.getName()), () -> { T val = (T) s.producer.sensors().get((AttributeSensor<?>) s.sensor); if (!Objects.equal(lastVal, val)) { if (LOG.isDebugEnabled()) LOG.debug("skipping initial value delivery of {} -> {} to {} as value changed from {} to {}", new Object[] {s.producer, s.sensor, s, lastVal, val}); return; } submitPublishEvent(s, new BasicSensorEvent<T>(s.sensor, s.producer, val), true); }); } return s; } LocalSubscriptionManager(ExecutionManager m); long getNumSubscriptions(); long getTotalEventsPublished(); long getTotalEventsDelivered(); @Override @SuppressWarnings("unchecked") Set<SubscriptionHandle> getSubscriptionsForSubscriber(Object subscriber); @Override synchronized Set<SubscriptionHandle> getSubscriptionsForEntitySensor(Entity source, Sensor<?> sensor); @Override @SuppressWarnings("rawtypes") synchronized boolean unsubscribe(SubscriptionHandle sh); @Override @SuppressWarnings({ "unchecked", "rawtypes" }) void publish(final SensorEvent<T> event); @Override String toString();  }
@Test public void testSubscribeToEntityAttributeChange() throws Exception { final CountDownLatch latch = new CountDownLatch(1); app.subscriptions().subscribe(entity, TestEntity.SEQUENCE, 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"); } }
@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"); } }
@Test public void testSubscribeToAttributeChangeWithEntityWildcard() throws Exception { final CountDownLatch latch = new CountDownLatch(1); app.subscriptions().subscribe(null, TestEntity.SEQUENCE, 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"); } }
@Test public void testSubscriptionReceivesInitialValueEventsInOrder() { RecordingSensorEventListener<Object> listener = new RecordingSensorEventListener<>(); entity.sensors().set(TestEntity.NAME, "myname"); entity.sensors().set(TestEntity.SEQUENCE, 123); entity.sensors().emit(TestEntity.MY_NOTIF, -1); entity.subscriptions().subscribe(ImmutableMap.of("notifyOfInitialValue", true), entity, TestEntity.SEQUENCE, listener); Asserts.succeedsEventually(() -> Asserts.assertSize(listener.getEvents(), 1)); entity.sensors().set(TestEntity.SEQUENCE, 456); entity.subscriptions().subscribe(ImmutableMap.of("notifyOfInitialValue", true), entity, TestEntity.MY_NOTIF, listener); entity.sensors().emit(TestEntity.MY_NOTIF, 1); entity.sensors().set(TestEntity.SERVICE_STATE_ACTUAL, Lifecycle.STOPPING); entity.subscriptions().subscribe(ImmutableMap.of("notifyOfInitialValue", true), entity, TestEntity.SERVICE_STATE_ACTUAL, listener); entity.subscriptions().subscribe(ImmutableMap.of("notifyOfInitialValue", true), entity, TestEntity.NAME, listener); Asserts.succeedsEventually(() -> { Asserts.assertEquals(listener.getEvents(), ImmutableList.of( new BasicSensorEvent<Integer>(TestEntity.SEQUENCE, entity, 123), new BasicSensorEvent<Integer>(TestEntity.SEQUENCE, entity, 456), new BasicSensorEvent<Integer>(TestEntity.MY_NOTIF, entity, 1), new BasicSensorEvent<Lifecycle>(TestEntity.SERVICE_STATE_ACTUAL, entity, Lifecycle.STOPPING), new BasicSensorEvent<String>(TestEntity.NAME, entity, "myname")), "actually got: "+listener.getEvents()); }); }
@Test public void testNotificationOrderMatchesSetValueOrderWhenSynched() { RecordingSensorEventListener<Object> listener = new RecordingSensorEventListener<>(); AtomicInteger count = new AtomicInteger(); Runnable set = () -> { synchronized (count) { entity.sensors().set(TestEntity.SEQUENCE, count.incrementAndGet()); } }; entity.subscriptions().subscribe(ImmutableMap.of(), entity, TestEntity.SEQUENCE, listener); for (int i=0; i<10; i++) { new Thread(set).start(); } Asserts.succeedsEventually(MutableMap.of("timeout", Duration.seconds(5)), () -> { Asserts.assertSize(listener.getEvents(), 10); }); for (int i=0; i<10; i++) { Assert.assertEquals(listener.getEvents().get(i).getValue(), i+1); } }
@Test public void testNotificationOrderMatchesSetValueOrderWhenNotSynched() { RecordingSensorEventListener<Object> listener = new RecordingSensorEventListener<>(); AtomicInteger count = new AtomicInteger(); Runnable set = () -> { entity.sensors().set(TestEntity.SEQUENCE, count.incrementAndGet()); }; entity.subscriptions().subscribe(ImmutableMap.of(), entity, TestEntity.SEQUENCE, listener); for (int i=0; i<10; i++) { new Thread(set).start(); } Asserts.succeedsEventually(MutableMap.of("timeout", Duration.seconds(5)), () -> { Asserts.assertSize(listener.getEvents(), 10); }); Assert.assertEquals(listener.getEvents().get(9).getValue(), entity.sensors().get(TestEntity.SEQUENCE)); }
@Test public void testSubscriptionHasSubscribersExecutionContext() throws Exception { final AtomicReference<BasicExecutionContext> result = new AtomicReference<>(); final CountDownLatch latch = new CountDownLatch(1); app.subscriptions().subscribe(entity, TestEntity.SEQUENCE, new SensorEventListener<Object>() { @Override public void onEvent(SensorEvent<Object> event) { result.set(BasicExecutionContext.getCurrentExecutionContext()); latch.countDown(); }}); entity.setSequenceValue(1234); if (!latch.await(TIMEOUT_MS, TimeUnit.MILLISECONDS)) { fail("Timeout waiting for Event on TestEntity listener"); } Assert.assertEquals(result.get(), app.getExecutionContext()); }