MhLabs.PubSubExtensions
5.0.0
dotnet add package MhLabs.PubSubExtensions --version 5.0.0
NuGet\Install-Package MhLabs.PubSubExtensions -Version 5.0.0
<PackageReference Include="MhLabs.PubSubExtensions" Version="5.0.0" />
paket add MhLabs.PubSubExtensions --version 5.0.0
#r "nuget: MhLabs.PubSubExtensions, 5.0.0"
// Install MhLabs.PubSubExtensions as a Cake Addin #addin nuget:?package=MhLabs.PubSubExtensions&version=5.0.0 // Install MhLabs.PubSubExtensions as a Cake Tool #tool nuget:?package=MhLabs.PubSubExtensions&version=5.0.0
MhLabs.PubSubExtensions
Extended functionality of AmazonSimpleNotificationServiceClient
and AmazonSQSClient
that handles messages larger than 256KB up to 2GB.
Known issues/limitations
- Only supports publishing to SNS. SQS will be a later feature
- Only supports SNS/SQS consumption through AWS Lambda using SNS or SQS as event source.
Usage
The package consists of two namespaces - Producer
and Consumer
.
Nuget: dotnet add package MhLabs.PubSubExtensions
Producer
To publish to an SNS topic, register the extended client to Startup.cs
:
services.AddSingleton<IAmazonSimpleNotificationService, ExtendedSimpleNotificationServiceClient>();
To publish a message:
public ProducingService(IAmazonSimpleNotificationService snsClient)
{
_snsClient = snsClient;
}
public async Task Publish(Model model)
{
var response = await _snsClient.PublishAsync(_topic, JsonConvert.SerializeObject(model));
}
In the above example model
could be on any size between 1 byte up to 2GB. The underlying logic will calculate the size of the stream and, if over 256KB, upload to S3 and setting MessageAttributes
of the S3 bucket and key. The consuming end will automatically download from S3 when appropriate.
Consumer
This is primarily designed to be consumed in an AWS Lambda function. Generally you want to publish to an SQS topic, subscribe an SQS queue to it and consume the queue in a Lambda. This allows you to consume the queue at the parallellism of your choice and it also provides built in retry logic.
Creating the subscription
In the Resources
section of serverless.template
:
"Topic": {
"Type": "AWS::SNS::Topic"
},
"Queue": {
"Type": "AWS::SQS::Queue"
},
"QueuePolicy": {
"Type": "AWS::SQS::QueuePolicy",
"Properties": {
"PolicyDocument": {
"Version": "2012-10-17",
"Id": "QueuePolicy",
"Statement": [
{
"Sid": "Allow-SendMessage-To-Both-Queues-From-SNS-Topic",
"Effect": "Allow",
"Principal": "*",
"Action": ["sqs:SendMessage"],
"Resource": "*",
"Condition": {
"ArnEquals": {
"aws:SourceArn": { "Ref": "Topic" }
}
}
}
]
},
"Queues": [{ "Ref": "Queue" }]
}
},
"Subscription": {
"Type": "AWS::SNS::Subscription",
"Properties": {
"TopicArn": {
"Ref": "Topic"
},
"Endpoint": {
"Fn::GetAtt": ["Queue", "Arn"]
},
"Protocol": "sqs",
"RawMessageDelivery": true
}
}
Also add a consumer Lambda resource to consume the queue:
"SQSConsumer": {
"Type": "AWS::Serverless::Function",
"Properties": {
"Handler": "example::example.Lambdas.SQSConsumer::Process",
"Runtime": "dotnetcore2.1",
"CodeUri": "bin/publish",
"MemorySize": 256,
"Timeout": 30,
"Role": null,
"Policies": ["AWSLambdaFullAccess", "AWSXrayWriteOnlyAccess"],
"Tracing": "Active",
"Environment": {
"Variables": {
"PubSub_S3Bucket": { "Ref": "Bucket" }
}
},
"Events": {
"SQS": {
"Type": "SQS",
"Properties": {
"Queue": { "Fn::GetAtt": ["Queue", "Arn"] },
"BatchSize": 5
}
}
}
}
}
Creating the consumer.
For SNS, SQS and Kinesis consumers, the message extraction and deserialization is performed on the base class. This to avoid boiletplate in your lambda handler.
public class SNSConsumer : SNSMessageMessageProcessorBase<Model>
{
protected override async Task HandleEvent(IEnumerable<Model> items, ILambdaContext context)
{
// Iterate through items
}
}
For SQS you have the option of using partial batch failures. If you do not want to use them there is no need to return any data. See example below.
public class SQSConsumerWithoutPartialBatchFailure : SQSMessageMessageProcessorBase<Model>
{
protected override async Task<SQSResponse> HandleEvent(IEnumerable<Model> items, ILambdaContext context)
{
// Iterate through items
// No need to return anything
return default;
}
}
When using partial batch failures you will need to return the message ids of the failed message in the response from the lambda.
public class SQSConsumerWithPartialBatchFailure : SQSMessageProcessorBase<Model>
{
// This tells the base class to use partial batch failure on exceptions as well
protected override Task<HandleErrorResult> HandleError(SQSEvent ev, ILambdaContext context, Exception exception)
{
return Task.FromResult(HandleErrorResult.ErrorHandledByConsumer);
}
protected override async Task<SQSResponse> HandleEvent(IEnumerable<SQSMessageEnvelope<Model>> items, ILambdaContext context)
{
var failures = new List<BatchItemFailure>();
foreach (var orderEvent in items)
{
try
{
// Do something with event
}
catch (System.Exception e)
{
// Add any failed messages to the batch failures
failures.Add(new BatchItemFailure
{
ItemIdentifier = orderEvent.MessageId
});
}
}
return new SQSResponse
{
BatchItemFailures = failures
};
}
}
Message extraction
If you want to perform more advanced message extraction, such at populate your model with values from MessageAttributes or so, you will have to create your own message extractor.
public class MyExtractor : IMessageExtractor
{
public Type ExtractorForType => typeof(SQSEvent);
public async Task<IEnumerable<TMessageType>> ExtractEventBody<TEventType, TMessageType>(TEventType ev)
{
var sqsEvent = ev as SQSEvent;
return await Task.Run(()=>sqsEvent.Records.Select(p => JsonConvert.DeserializeObject<TMessageType>(p.MessageAttributes["SomeAttributeWithJsonBody"].StringValue)));
}
}
and register it with the base:
public class SQSConsumer : MessageProcessorBase<SQSEvent, Model>
{
public SQSConsumer() {
base.RegisterExtractor(new MyExtractor());
}
protected override async Task HandleEvent(IEnumerable<Model> items, ILambdaContext context)
{
// Iterate through items
}
}
Pushing a new version
Set the Version
number in the <a href="https://github.com/mhlabs/MhLabs.PubSubExtensions/blob/master/MhLabs.PubSubExtensions/MhLabs.PubSubExtensions.csproj"> .csproj-file</a> before pushing. If an existing version is pushed the <a href="https://github.com/mhlabs/MhLabs.PubSubExtensions/actions">build will fail</a>.
Publish pre-release packages on branches to allow us to test the package without merging to master
- Create a new branch
- Update
Version
number and add-beta
postfix (can have .1, .2 etc. at the end) - Make any required changes updating the version as you go
- Test beta package in solution that uses package
- Create PR and get it reviewed
- Check if there are any changes on the branch you're merging into. If there are you need to rebase those changes into yours and check that it still builds
- As the final thing before merging update version number and remove post fix
Product | Versions Compatible and additional computed target framework versions. |
---|---|
.NET | net5.0 was computed. net5.0-windows was computed. net6.0 was computed. net6.0-android was computed. net6.0-ios was computed. net6.0-maccatalyst was computed. net6.0-macos was computed. net6.0-tvos was computed. net6.0-windows was computed. net7.0 was computed. net7.0-android was computed. net7.0-ios was computed. net7.0-maccatalyst was computed. net7.0-macos was computed. net7.0-tvos was computed. net7.0-windows was computed. net8.0 was computed. net8.0-android was computed. net8.0-browser was computed. net8.0-ios was computed. net8.0-maccatalyst was computed. net8.0-macos was computed. net8.0-tvos was computed. net8.0-windows was computed. |
.NET Core | netcoreapp3.0 was computed. netcoreapp3.1 was computed. |
.NET Standard | netstandard2.1 is compatible. |
MonoAndroid | monoandroid was computed. |
MonoMac | monomac was computed. |
MonoTouch | monotouch was computed. |
Tizen | tizen60 was computed. |
Xamarin.iOS | xamarinios was computed. |
Xamarin.Mac | xamarinmac was computed. |
Xamarin.TVOS | xamarintvos was computed. |
Xamarin.WatchOS | xamarinwatchos was computed. |
-
.NETStandard 2.1
- Amazon.Lambda.Core (>= 2.1.0)
- Amazon.Lambda.DynamoDBEvents (>= 2.1.1)
- Amazon.Lambda.KinesisEvents (>= 2.0.0)
- Amazon.Lambda.Serialization.Json (>= 2.1.0)
- Amazon.Lambda.SNSEvents (>= 2.0.0)
- Amazon.Lambda.SQSEvents (>= 2.1.0)
- AWSSDK.S3 (>= 3.7.101.53)
- AWSSDK.SimpleNotificationService (>= 3.7.100.54)
- AWSSDK.SQS (>= 3.7.100.53)
- AWSSDK.StepFunctions (>= 3.7.102.26)
- Microsoft.Extensions.Logging (>= 7.0.0)
- Newtonsoft.Json (>= 13.0.2)
NuGet packages
This package is not used by any NuGet packages.
GitHub repositories
This package is not used by any popular GitHub repositories.
Version | Downloads | Last updated |
---|---|---|
5.0.0 | 7,975 | 1/13/2023 |
4.1.0 | 2,300 | 7/20/2022 |
4.0.0 | 3,347 | 1/19/2022 |
3.0.0 | 3,763 | 10/21/2021 |
2.0.0 | 7,669 | 1/12/2021 |
1.0.56 | 13,297 | 5/18/2020 |
1.0.53 | 6,611 | 2/14/2020 |
1.0.52 | 1,250 | 1/28/2020 |
1.0.51 | 3,712 | 11/17/2019 |
1.0.50 | 545 | 11/16/2019 |
1.0.49 | 724 | 11/14/2019 |
1.0.48 | 537 | 11/14/2019 |
1.0.47 | 560 | 11/13/2019 |
1.0.46 | 557 | 11/13/2019 |
1.0.45 | 551 | 11/13/2019 |
1.0.44 | 1,793 | 11/5/2019 |
1.0.43 | 3,732 | 7/26/2019 |
1.0.42 | 575 | 7/25/2019 |
1.0.41 | 1,988 | 5/2/2019 |
1.0.40 | 1,199 | 4/11/2019 |
1.0.33 | 661 | 4/10/2019 |
1.0.32 | 623 | 4/8/2019 |
1.0.31 | 608 | 4/8/2019 |
1.0.30 | 1,548 | 4/1/2019 |
1.0.29 | 2,311 | 3/8/2019 |
1.0.28 | 649 | 3/8/2019 |
1.0.27 | 2,353 | 2/28/2019 |
1.0.26 | 610 | 2/28/2019 |
1.0.25 | 611 | 2/28/2019 |
1.0.24-beta | 636 | 2/6/2019 |
1.0.23 | 1,257 | 2/4/2019 |
1.0.22 | 681 | 2/4/2019 |
1.0.21 | 689 | 2/4/2019 |
1.0.20 | 691 | 2/4/2019 |
1.0.19 | 683 | 2/4/2019 |
1.0.18 | 969 | 2/4/2019 |
1.0.17 | 661 | 1/31/2019 |
1.0.16 | 677 | 1/31/2019 |
1.0.15 | 676 | 1/31/2019 |
1.0.14 | 674 | 1/31/2019 |
1.0.14-beta | 574 | 1/30/2019 |
1.0.13-beta | 534 | 1/30/2019 |
1.0.12 | 726 | 1/29/2019 |
1.0.11 | 1,243 | 1/21/2019 |
1.0.10 | 663 | 1/21/2019 |
1.0.9 | 2,208 | 1/2/2019 |
1.0.8 | 1,974 | 11/5/2018 |
1.0.7 | 928 | 10/31/2018 |
1.0.6 | 716 | 10/31/2018 |
1.0.5 | 913 | 10/30/2018 |
1.0.4 | 723 | 10/29/2018 |
1.0.3 | 868 | 10/24/2018 |
1.0.2 | 1,197 | 10/10/2018 |
1.0.1 | 857 | 10/9/2018 |
1.0.0 | 3,761 | 10/4/2018 |