# Message Throughput Throttling


Systems often need to integrate with 3rd party services, some of which may limit the number of concurrent requests they process.

This sample demonstrates an integration with the [GitHub API](https://developer.github.com/v3/) using the [Octokit library](https://github.com/octokit/octokit.net). It runs in unauthenticated mode, which is limited to a fixed number of requests per minute. Upon hitting this limit, the endpoint will delay processing additional messages, until the limit resets.

The solution consists of two endpoints; Sender and Limited. The reason two endpoints are required in this scenario is that NServiceBus does not support limiting messages by message type. So, to limit **only** a specific message type, a separate endpoint is used for it.

## Sender

The Sender is a normal endpoint that sends the `SearchGitHub` message and then handles the reply `SearchResponse` message.

### Sending

The message sending occurs at startup.

<!-- snippet: Sending -->

```cs
Console.WriteLine("Sending messages...");
for (var i = 0; i < 100; i++)
{
    var searchGitHub = new SearchGitHub
    {
        Repository = "NServiceBus",
        Owner = "Particular",
        Branch = "master"
    };
    await messageSession.Send("Samples.Throttling.Limited", searchGitHub);
}
```

<!-- endsnippet -->

### Handling the response

Handling the GitHubSearchResponse message.

<!-- snippet: GitHubSearchResponseHandler -->

```cs
public class GitHubSearchResponseHandler(ILogger<GitHubSearchResponseHandler> logger) :
    IHandleMessages<SearchResponse>
{
    public Task Handle(SearchResponse message, IMessageHandlerContext context)
    {
        logger.LogInformation("Found commit '{CommitSha}' for branch '{Branch}.'", message.CommitSha, message.Branch);
        return Task.CompletedTask;
    }
}
```

<!-- endsnippet -->

## Limited

This endpoint is limited to processing one message at a time and uses a [pipeline behavior](/nservicebus/pipeline/manipulate-with-behaviors.md) to handle the processing limit of Octokit being exceeded.

### Configure to process only one concurrent message

[Limits the endpoint concurrency](/nservicebus/operations/tuning.md).

<!-- snippet: LimitConcurrency -->

```cs
var endpointConfiguration = new EndpointConfiguration("Samples.Throttling.Limited");
endpointConfiguration.LimitMessageProcessingConcurrencyTo(1);
```

<!-- endsnippet -->

### Search Handler

Performs the Octokit search.

<!-- snippet: SearchHandler -->

```cs
public class GitHubSearchHandler(ILogger<GitHubSearchHandler> logger) :
    IHandleMessages<SearchGitHub>
{

    // use anonymous access which has strict rate limitations
    GitHubClient GitHubClient = new GitHubClient(new ProductHeaderValue("ThroughputThrottling"));

    public async Task Handle(SearchGitHub message, IMessageHandlerContext context)
    {
        logger.LogInformation("Received search request for branch '{Branch}' on '{Owner}/{Repository}'", message.Branch, message.Owner, message.Repository);

        var result = await GitHubClient.Repository.Branch.Get(message.Owner, message.Repository, "master");
        logger.LogInformation("Found commit '{CommitSha}' for branch '{Branch}'. Replying.", result.Commit.Sha, message.Branch);
        var response = new SearchResponse
        {
            Branch = message.Branch,
            CommitSha = result.Commit.Sha
        };
        await context.Reply(response);
    }
}
```

<!-- endsnippet -->

### Registering the behavior in pipeline

<!-- snippet: RegisterBehavior -->

```cs
endpointConfiguration.Pipeline.Register(typeof(ThrottlingBehavior), "API throttling for GitHub");
```

<!-- endsnippet -->

### The pipeline behavior

Handles the detection of `Octokit.RateLimitExceededException` and defers the message.

<!-- snippet: ThrottlingBehavior -->

```cs
public class ThrottlingBehavior(ILogger<ThrottlingBehavior> logger) :
    Behavior<IInvokeHandlerContext>
{
    static DateTime? nextRateLimitReset;

    public override async Task Invoke(IInvokeHandlerContext context, Func<Task> next)
    {
        var rateLimitReset = nextRateLimitReset;
        if (rateLimitReset.HasValue && rateLimitReset >= DateTime.UtcNow)
        {
            var localTime = rateLimitReset?.ToLocalTime();
            logger.LogInformation("Rate limit exceeded. Retry after {RateLimitReset} UTC ({LocalTime} local).", rateLimitReset, localTime);
            await DelayMessage(context, rateLimitReset.Value);
            return;
        }

        try
        {
            await next();
        }
        catch (RateLimitExceededException exception)
        {
            var nextReset = nextRateLimitReset = exception.Reset.UtcDateTime;
            var localTime = nextReset?.ToLocalTime();
            logger.LogInformation("Rate limit exceeded. Limit resets at {NextReset} UTC ({LocalTime} local).", nextReset, localTime);
            await DelayMessage(context, nextReset.Value);
        }
    }

    Task DelayMessage(IInvokeHandlerContext context, DateTime deliverAt)
    {
        var sendOptions = new SendOptions();

        // delay the message to the specified delivery date
        sendOptions.DoNotDeliverBefore(deliverAt);
        // send message to this endpoint
        sendOptions.RouteToThisEndpoint();
        // maintain the original ReplyTo address
        if (context.Headers.TryGetValue(Headers.ReplyToAddress, out var replyAddress))
        {
            sendOptions.RouteReplyTo(replyAddress);
        }

        return context.Send(context.MessageBeingHandled, sendOptions);
    }
}
```

<!-- endsnippet -->

> [!NOTE]
> The behavior sends a *copy* of the original message, but does not copy the headers of the original message. If the headers of the original message are required, they must be copied from `context.Headers` to `SendOptions`.