Polling 2.0

Stories

1. As a user, I want to execute a given flow every hour regardless of the day/week on Monday, Weds, Friday at 12 AM Every 5 minutes
2. As a user, I want to be able to try out my poll: a. I do not have to wait for the next poll to execute (which may be a day away) to test what will happen b. I do not get confused about what is happening because polls continue to execute while I'm trying to debug what is happening
3. As a user, I want to be able to test my poll: a. I want to run a unit test that executes a poll b. I want to start a test without having polls running.
4. As an extension developer, I want better APIs to start, stop and execute polls.
5. As a user, I want to prevent poll executions from executing concurrently
6. As an extension developer I want to change the polling strategy on runtime
7. As an extension developer I want to start/stop/run my polls
8. As an extension developer I want to enrich with notifications of the poll execution.

Proposals

Mule API to support extension developer's features.

Nowadays we don't have a mule java API in order to manage the polls, this implies that are features that we cannot perform or are hard to develop.

Polling Manager and Scheduler

We could have a poll manager. This will handle all types of polling (in the future, for now we can make it handle the poll elements only)

Definition

/**
 * Handles the polls in the application.
 */
public final class PollingManager implements Startable, Stoppable, MuleContextAware
{

    protected MuleContext muleContext;

    /**
     * <p>
     *     Adds a new polling scheduler
     * </p>
     */
    public void register(PollScheduler scheduler);

    /**
     * <p>
     *     Removes the schedulers that matches comparator equals
     * </p>
     */
    public void unRegister(Predicate<PollSchedulerId> comparator);

    /**
     * <p>
     *     Retrieves a set of schedulers based on the comparator matcher
     * </p>
     */
    public  Collection<PollScheduler> lookupSchedulers(Predicate<PollSchedulerId> comparator);

    @Override
    public void start() throws MuleException
    {
        muleContext.fireNotification(new PollingManagerNotification(START_BEGIN, this));

        // Starts the schedulers
        doStart();
        
        muleContext.fireNotification(new PollingManagerNotification(START_END, this));
    }


    @Override
    public void stop() throws MuleException
    {
        muleContext.fireNotification(new PollingManagerNotification(STOP_BEGIN, this));

        // Stops the schedulers
        doStop();

        muleContext.fireNotification(new PollingManagerNotification(STOP_END, this));
    }


    public void setMuleContext(MuleContext context)
    {
       this.muleContext = context; 
    }
}

The Scheduler knows when to execute the receiver and for whom.

{code:java} public interface PollScheduler extends Stoppable, Startable { /** *

* Scheduler identifier *

*/ PollSchedulerId getId();

/**
 * <p>
 *     Runs a scheduler on demand
 * </p>
 */
void run() throws MuleException;

/**
 * <p>
 *     Runs a scheduler on demand with a mule event
 * </p>
 */
void run(MuleEvent event) throws MuleException;

} {code}

The PollSchedulerId defines the Whom of the scheduler (the flow and endpoint address defines unicity)

{code:java} public class PollSchedulerId {

protected String flowConstructName;
protected EndpointURI endpointURI;

public PollSchedulerId(String flowConstructName, EndpointURI endpointURI)
{
    this.flowConstructName = flowConstructName;
    this.endpointURI = endpointURI;
}

public String getFlowConstructName()
{
    return flowConstructName;
}

public EndpointURI getEndpointURI()
{
    return endpointURI;
}

} {code}

h4. Usage

h5. Starting/running/stopping polls.

Covered scenarios:

1. I do not have to wait for the next poll to execute (which may be a day away) to test what will happen
2. As an extension developer, I want better APIs to start, stop and execute polls.
3. As an extension developer I want to change the polling strategy on runtime
4. As an extension user I want to start/stop/run my polls

{code:java}

public class MyPollingService extends AbstractService implements PollingService {

public void executePoll(String appName, final String flowName) throws MuleException { MuleContext context = getContext(appName); PollingManager pollingManager = context.getPollingManager();

    Collection<PollScheduler> schedulers = pollingManager.lookupSchedulers(new Predicate<PollSchedulerId>()
    {
        @Override
        public boolean evaluate(PollSchedulerId id)
        {
            return id.getFlowConstructName().equalsIgnoreCase(flowName);
        }
    });
    
    for (PollScheduler scheduler : schedulers){
        scheduler.run();
    }

}

}

{code}

h5. Starting with the polls disabled

1. I do not get confused about what is happening because polls continue to execute while I'm trying to debug what is happening
2. I want to start a test without having polls running.
3. As an extension developer I want to change the polling strategy on runtime
4. As an extension developer I want to enrich with notifications of the poll execution.

{code:java} public class BoostrapExtension implements Initialisable, MuleContextAware {

private MuleContext muleContext;

@Override
public void initialise() throws InitialisationException
{
    muleContext.getNotificationManager().addListener(new ServerNotificationListener<PollingManagerNotification>(){

        @Override
        public void onNotification(PollingManagerNotification notification)
        {
            if ( isStartBeginOfPollingManager(notification) ){
               PollingManager manager = notification.getSource();

                Collection<PollScheduler> schedulers = manager.lookupSchedulers(new Predicate<PollSchedulerId>()
                {
                    @Override
                    public boolean evaluate(PollSchedulerId id)
                    {
                        String path = id.getEndpointURI().getPath();

                        return path.startsWith("polling://");
                    }
                });
                
                for (PollScheduler scheduler : schedulers ){
                    manager.register(new MyScheduler(scheduler.getId()));
                }

            }
        }
    });
}

@Override
public void setMuleContext(MuleContext context)
{
    this.muleContext = context;
}

} {code}

h2. XML definition for final users

The new implementation of poll allows users to inject the polling strategy.

h3. The default implementation to support backward compatibility:

{code:xml}

{code}

h3. Injecting polling strategy by using the quartz module:

Now the quartz module will allow users to define a polling strategy that can be used by poll, and deprecate the quartz inbound endpoint.

{code:xml}

<quartz:polling-strategy name="myPollingStrategy" cronExpression="" />

{code}

h3. Defining your own polling strategy

Users can create a bean that defines the polling strategy:

{code:xml}

<spring:bean name="customPollingStrategy" class="org.mule.mine.MyPollingStrategyDefinition"> </spring:bean>

{code} And the user can define his own scheduler {code:java}

public class MyPollingStrategyDefinition implements PollSchedulerBuilder { @Override public PollScheduler buildFor(AbstractPollingMessageReceiver receiver) { return createPollScheduler(); }

} {code}