MhLabs.PubSubExtensions 5.0.0

dotnet add package MhLabs.PubSubExtensions --version 5.0.0                
NuGet\Install-Package MhLabs.PubSubExtensions -Version 5.0.0                
This command is intended to be used within the Package Manager Console in Visual Studio, as it uses the NuGet module's version of Install-Package.
<PackageReference Include="MhLabs.PubSubExtensions" Version="5.0.0" />                
For projects that support PackageReference, copy this XML node into the project file to reference the package.
paket add MhLabs.PubSubExtensions --version 5.0.0                
#r "nuget: MhLabs.PubSubExtensions, 5.0.0"                
#r directive can be used in F# Interactive and Polyglot Notebooks. Copy this into the interactive tool or source code of the script to reference the package.
// 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

  1. Create a new branch
  2. Update Version number and add -beta postfix (can have .1, .2 etc. at the end)
  3. Make any required changes updating the version as you go
  4. Test beta package in solution that uses package
  5. Create PR and get it reviewed
  6. 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
  7. As the final thing before merging update version number and remove post fix
Product 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. 
Compatible target framework(s)
Included target framework(s) (in package)
Learn more about Target Frameworks and .NET Standard.

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,961 1/13/2023
4.1.0 2,293 7/20/2022
4.0.0 3,338 1/19/2022
3.0.0 3,756 10/21/2021
2.0.0 7,658 1/12/2021
1.0.56 13,287 5/18/2020
1.0.53 6,600 2/14/2020
1.0.52 1,241 1/28/2020
1.0.51 3,703 11/17/2019
1.0.50 536 11/16/2019
1.0.49 716 11/14/2019
1.0.48 529 11/14/2019
1.0.47 552 11/13/2019
1.0.46 547 11/13/2019
1.0.45 543 11/13/2019
1.0.44 1,783 11/5/2019
1.0.43 3,724 7/26/2019
1.0.42 565 7/25/2019
1.0.41 1,978 5/2/2019
1.0.40 1,189 4/11/2019
1.0.33 651 4/10/2019
1.0.32 613 4/8/2019
1.0.31 600 4/8/2019
1.0.30 1,540 4/1/2019
1.0.29 2,301 3/8/2019
1.0.28 637 3/8/2019
1.0.27 2,345 2/28/2019
1.0.26 602 2/28/2019
1.0.25 603 2/28/2019
1.0.24-beta 632 2/6/2019
1.0.23 1,249 2/4/2019
1.0.22 673 2/4/2019
1.0.21 681 2/4/2019
1.0.20 683 2/4/2019
1.0.19 675 2/4/2019
1.0.18 959 2/4/2019
1.0.17 653 1/31/2019
1.0.16 667 1/31/2019
1.0.15 666 1/31/2019
1.0.14 666 1/31/2019
1.0.14-beta 570 1/30/2019
1.0.13-beta 528 1/30/2019
1.0.12 716 1/29/2019
1.0.11 1,235 1/21/2019
1.0.10 653 1/21/2019
1.0.9 2,200 1/2/2019
1.0.8 1,961 11/5/2018
1.0.7 915 10/31/2018
1.0.6 699 10/31/2018
1.0.5 900 10/30/2018
1.0.4 708 10/29/2018
1.0.3 855 10/24/2018
1.0.2 1,183 10/10/2018
1.0.1 837 10/9/2018
1.0.0 3,746 10/4/2018