---
schema_version: "1"
id: 8e976b6ac1a13fe28ded12d2fcc893098e9df76956d85af605a89153b4475be7
title: Polling Sources
collection: mulesoft
version: current
locale: en-us
portal_url: https://docs.salesforce.com/mulesoft/current/en-us/mule-sdk/latest/polling-sources.md
canonical_url: https://docs.mulesoft.com/mule-sdk/latest/polling-sources
source_site: docs.mulesoft.com
content_type: documentation
content_hash: 4114c075341817110604519878ff3e04985bcdb2c3a28d2ef3f56b353955f54e
description: MuleSoft Documentation Site
release: "1.13"
latest: true
---
---
product: Mule SDK
version: 1.13
is-latest-version: true
---

# Polling Sources

> For the full documentation index, see: https://docs.mulesoft.com/llms.txt

**Available since version 1.1**

Instead of getting triggered by an external action, Polling Sources are sources that poll items
periodically depending on a given `scheduling strategy` and automatically handle idempotency and
watermarking. On every poll action, many items can be dispatched to the flow, each as a single
`Message`.

## Creating a Polling Source

To create a Polling Source, you need to add the `Source` class to the `@Sources` annotation on the
extension class, just like regular Sources. For a Source to behave as a `PollingSource`, instead of
extending the class `Source<T,A>`, the class needs to extend from `PollingSource<T, A>`.

When extending from `PollingSource<T, A>`, you need to implement the methods `poll` and
`onRejectedItem`.

- `poll` is responsible for obtaining the items to be dispatched and communicating with them through
  the `pollContext` using the `accept` method.

- `onRejectedItem` is called when one of the dispatch items was rejected (for example, by
  watermarking idempotency or server overload). This method is called to release any resources
  associated with the result set for the `PollItem`.

Here is an example of the declaration of these methods on the directory listener implementation for
the FTP connector:

```java
public class FtpDirectoryListener extends PollingSource<InputStream, FtpFileAttributes> {
// ...
  @Override
  public void poll(PollContext<InputStream, FtpFileAttributes> pollContext) {
    // ...
  }

  @Override
  public void onRejectedItem(Result<InputStream, FtpFileAttributes> result, SourceCallbackContext callbackContext) {
    // ...
  }
// ...
}
```

## Implementing the Poll Method

This method gathers all the items to be dispatched and processes them one by one.

The `accept` method from the `PollContext` is in charge of dispatching the polled items into the
flow. It receives `Consumer<PollItem>` as a parameter.

This example shows how to set the Message to dispatch to the flow:

```java
public class FtpDirectoryListener extends PollingSource<InputStream, FtpFileAttributes> {
// ...
  @Override
  public void poll(PollContext<InputStream, FtpFileAttributes> pollContext) {
    if(pollContext.isSourceStopping()){ (1)
      return;
    }
    List<FtpFileAttributes> attributesList = listFilesAttributes();

    for (FtpFileAttributes attributes : attributesList) {
      if(pollContext.isSourceStopping()){ (1)
        break;
      }
      pollContext.accept(item -> { (2)
        Result<InputStream, FtpFileAttributes> result = read(attributes.getPath());
        item.setResult(result); (3)
      });
    }
  }
// ...
}
```

<table><tbody><tr><td><i data-value="1"></i><b>1</b></td><td>Note that the <code>pollContext</code> provides information on whether the Source is stopping. This must be taken into account before any time-consuming task.</td></tr><tr><td><i data-value="2"></i><b>2</b></td><td><code>accept</code> dispatches an item to the flow.</td></tr><tr><td><i data-value="3"></i><b>3</b></td><td><code>setResult</code> is used to set the Message that is dispatched.</td></tr></tbody></table>

### Idempotency

Idempotency prevents items from getting polled twice. The SDK ensures that no other thread or
cluster node is processing an item of the same ID, which provides processing idempotency. No two
items with the same ID will be processed at the same time. If a node polls an item with the ID of an
item that is being processed, the item will be dropped, and the `onRejectedItem` method will be
called.

```java
public class FtpDirectoryListener extends PollingSource<InputStream, FtpFileAttributes> {
// ...
  @Override
  public void poll(PollContext<InputStream, FtpFileAttributes> pollContext) {
    if(pollContext.isSourceStopping()){
      return;
    }
    List<FtpFileAttributes> attributesList = listFilesAttributes();

    for (FtpFileAttributes attributes : attributesList) {
      if(pollContext.isSourceStopping()){
        break;
      }
      pollContext.accept(item -> {
        Result<InputStream, FtpFileAttributes> result = read(attributes.getPath());
        item.setResult(result);
        item.setId(attributes.getPath());  (1)
      });
    }
  }
// ...
}
```

<table><tbody><tr><td><i data-value="1"></i><b>1</b></td><td><code>setId</code> is used to set the ID used for idempotency.</td></tr></tbody></table>

### Watermarking

You can use watermarking when you are polling an incremental source of data and you only want the
new data instead of manually dividing the new data from the old data on each call.

By setting an incrementing value to the item’s watermark, every other poll will drop and call
`onRejectedItem` on any of the item’s watermark values that are lower than the current watermark
value. In the example above, the chosen watermark is the timestamp of the file.

What criteria does the SDK use to compare watermarks?

If the watermarks that you use implement `Comparable`, the natural order of those values will be
used. There is no need to set a criterion.

```java
public class FtpDirectoryListener extends PollingSource<InputStream, FtpFileAttributes> {
// ...
  @Parameter
  @Optional(defaultValue = "false")
  private boolean watermarkEnabled = false; (1)
// ...
  @Override
  public void poll(PollContext<InputStream, FtpFileAttributes> pollContext) {
    if(pollContext.isSourceStopping()){
      return;
    }
    List<FtpFileAttributes> attributesList = listFilesAttributes();

    for (FtpFileAttributes attributes : attributesList) {
      if(pollContext.isSourceStopping()){
        break;
      }
      pollContext.accept(item -> {
        Result<InputStream, FtpFileAttributes> result = read(attributes.getPath());
        item.setResult(result);
        item.setId(attributes.getPath());
        if (watermarkEnabled) { (1)
          item.setWatermark(attributes.getTimestamp());  (2)
        }
      });
    }
  }
// ...
}
```

<table><tbody><tr><td><i data-value="1"></i><b>1</b></td><td>A typical best practice is to let the user decide whether to use watermarking or not.</td></tr><tr><td><i data-value="2"></i><b>2</b></td><td><code>setWatermark</code> sets the file timestamp as watermark.</td></tr></tbody></table>

What if the watermark values do not implement `Comparable`, or what if I want to use another
criteria that is not the natural order?

You can simply set a `Comparator` to the `PollContext` by calling its method
`setWatermarkComparator`.

> [!NOTE] If the watermark updates are not set before polling is complete, you can add a latch to
> avoid updating the watermark in two iterations.

### PollItemStatus

If you call the `accept` method, you get a `PollItemStatus` in return. This is an `Enum` that can
take the following values:

- ACCEPTED: The item was accepted and has been scheduled for execution.

- FILTERED\_BY\_WATERMARK: The item was rejected because watermarking was enabled, so it was
  filtered.

- ALREADY\_IN\_PROCESS: The item was rejected because idempotency was enabled and another thread or
  node is already processing this item.

- SOURCE\_STOPPING: The item was rejected because the source has received the stop signal.

You could use this to log the status of each item polled:

```java
public class FtpDirectoryListener extends PollingSource<InputStream, FtpFileAttributes> {
// ...
  @Parameter
  @Optional(defaultValue = "false")
  private boolean watermarkEnabled = false;
// ...
  @Override
  public void poll(PollContext<InputStream, FtpFileAttributes> pollContext) {
    if(pollContext.isSourceStopping()){
      return;
    }
    List<FtpFileAttributes> attributesList = listFilesAttributes();

    for (FtpFileAttributes attributes : attributesList) {
      if(pollContext.isSourceStopping()){
        break;
      }
      PollItemStatus status = pollContext.accept(item -> { (1)
        Result<InputStream, FtpFileAttributes> result = read(attributes.getPath());
        item.setResult(result);
        item.setId(attributes.getPath());
        if (watermarkEnabled) {
          item.setWatermark(attributes.getTimestamp());
        }
      });

      if(!status.equals(PollItemStatus.ACCEPTED)){
        LOGGER.debug("Item rejected with code:  " + status.name());
      }
    }
  }
// ...
}
```

<table><tbody><tr><td><i data-value="1"></i><b>1</b></td><td>Get the <code>PollItemStatus</code> that is logged later.</td></tr></tbody></table>

### Getting the Source Callback Context

If you need to save data to be used on the callback methods, `item` provides the
`SourceCallbackContext` for it.

This example shows how to get the context and then save data in it:

```java
public class FtpDirectoryListener extends PollingSource<InputStream, FtpFileAttributes> {
// ...
  @Parameter
  @Optional(defaultValue = "false")
  private boolean watermarkEnabled = false;
// ...
  @Override
  public void poll(PollContext<InputStream, FtpFileAttributes> pollContext) {
    if(pollContext.isSourceStopping()){
      return;
    }
    List<FtpFileAttributes> attributesList = listFilesAttributes();

    for (FtpFileAttributes attributes : attributesList) {
      if(pollContext.isSourceStopping()){
        break;
      }
      PollItemStatus status = pollContext.accept(item -> {
        SourceCallbackContext context = item.getSourceCallbackContext();   (1)
        context.addVariable("attributes", attributes);  (2)
        Result<InputStream, FtpFileAttributes> result = read(attributes.getPath());
        item.setResult(result);
        item.setId(attributes.getPath());
        if (watermarkEnabled) {
          item.setWatermark(attributes.getTimestamp());
        }
      });

      if(!status.equals(PollItemStatus.ACCEPTED)){
        LOGGER.debug("Item rejected with code:  " + status.name());
      }
    }
  }
// ...
}
```

<table><tbody><tr><td><i data-value="1"></i><b>1</b></td><td>Gets the context from <code>item</code>.</td></tr><tr><td><i data-value="2"></i><b>2</b></td><td>Saves data on the <code>SourceCallbackContext</code>.</td></tr></tbody></table>

## Implementing the OnRejectedItem Method

This method is called when one of the dispatched items was rejected. It should release any resource
associated with the result set for the `PollItem`.

This example shows how to implement it:

```java
public class FtpDirectoryListener extends PollingSource<InputStream, FtpFileAttributes> {
// ...
  @Override
    public void onRejectedItem(Result<InputStream, FtpFileAttributes> result, SourceCallbackContext callbackContext) {
      closeFileStream(result);
    }
// ...
}
```

## How to Use Polling Sources in a Flow

The SDK automatically adds a `scheduling strategy` parameter to the source, enabling the user to
provide any strategy (like `fixed-frequency`) to handle the polling executions. The runtime will use
that strategy to automatically schedule executions of the `poll` method.

Here is an example of the FTP directory listener that uses the polling source:

```xml
<ftp:listener config-ref="config" directory="path/" watermarkEnabled="true">
  <scheduling-strategy>
    <fixed-frequency startDelay="40000" frequency="1000" timeUnit="MILLISECONDS"/>
  </scheduling-strategy>
</ftp:listener>
```