Poller { @SuppressWarnings({ "unchecked" }) public void start() { if (log.isDebugEnabled()) log.debug("Starting poll for {} (using {})", new Object[] {entity, this}); if (started) { throw new IllegalStateException(String.format("Attempt to start poller %s of entity %s when already running", this, entity)); } started = true; for (final Callable<?> oneOffJob : oneOffJobs) { Task<?> task = Tasks.builder().dynamic(false).body((Callable<Object>) oneOffJob).displayName("Poll").description("One-time poll job "+oneOffJob).build(); oneOffTasks.add(feed.getExecutionContext().submit(task)); } Duration minPeriod = null; for (final PollJob<V> pollJob : pollJobs) { final String scheduleName = pollJob.handler.getDescription(); if (pollJob.pollPeriod.compareTo(Duration.ZERO) > 0) { ScheduledTask t = ScheduledTask.builder(() -> { DynamicSequentialTask<Void> task = new DynamicSequentialTask<Void>(MutableMap.of("displayName", scheduleName, "entity", entity), new Callable<Void>() { @Override public Void call() { if (!Entities.isManaged(entity)) { return null; } if (onlyIfServiceUp && !Boolean.TRUE.equals(entity.getAttribute(Attributes.SERVICE_UP))) { return null; } pollJob.wrappedJob.run(); return null; } } ); BrooklynTaskTags.setTransient(task); return task; }) .displayName("scheduled:" + scheduleName) .period(pollJob.pollPeriod) .cancelOnException(false) .build(); tasks.add(Entities.submit(entity, t)); if (minPeriod==null || (pollJob.pollPeriod.isShorterThan(minPeriod))) { minPeriod = pollJob.pollPeriod; } } else { if (log.isDebugEnabled()) log.debug("Activating poll (but leaving off, as period {}) for {} (using {})", new Object[] {pollJob.pollPeriod, entity, this}); } } if (minPeriod!=null && feed!=null) { feed.highlightTriggerPeriod(minPeriod); } } @Deprecated  Poller(Entity entity, boolean onlyIfServiceUp);  Poller(Entity entity, AbstractFeed feed, boolean onlyIfServiceUp); void submit(Callable<?> job); void scheduleAtFixedRate(Callable<V> job, PollHandler<? super V> handler, long period); void scheduleAtFixedRate(Callable<V> job, PollHandler<? super V> handler, Duration period); @SuppressWarnings({ "unchecked" }) void start(); void stop(); boolean isRunning(); @Override String toString(); static final Logger log; }
@Test(dataProvider = "specProvider", groups={"Broken"}) public void testFeedContinuesWhenPollerThrows(EntitySpec<FeedExceptionEntity> spec) { Map<?, ?> timeoutFlags = ImmutableMap.of("timeout", "100ms"); FeedExceptionEntity fee = app.createAndManageChild(spec); app.start(ImmutableList.of(app.newSimulatedLocation())); EntityAsserts.assertAttributeEqualsEventually(timeoutFlags, fee, FeedExceptionEntity.FLAG, true); fee.startThrowingPollExceptions(); EntityAsserts.assertAttributeEqualsEventually(timeoutFlags, fee, FeedExceptionEntity.FLAG, false); EntityAsserts.assertAttributeEqualsContinually(timeoutFlags, fee, FeedExceptionEntity.FLAG, false); fee.stopThrowingPollExceptions(); EntityAsserts.assertAttributeEqualsEventually(timeoutFlags, fee, FeedExceptionEntity.FLAG, true); EntityAsserts.assertAttributeEqualsContinually(timeoutFlags, fee, FeedExceptionEntity.FLAG, true); }