Adding messages to IAsyncCollector Topic output wi

Question

Is it currently possible to push messages to an IAsyncCollector Topic output from Azure functions and also set a session ID? My topics really on FIFO ordering and so we had to setup sessions. Because of this we had imagined just setting up a Guid to be the unique session id. I know how I would push messages to my topic through this output but of course that errors out since we aren't setting the session Id explicitly. Is it possible to set this somewhere in the code as we send it off to the IAsyncCollector?

Here is what we have,

[FunctionName("AccountCreatedHook")]
public static async Task<HttpResponseMessage> Run([HttpTrigger(AuthorizationLevel.Function, "post", Route = null)]HttpRequestMessage req,
    TraceWriter log, [ServiceBus("topic-name", Connection = "busname", EntityType = Microsoft.Azure.WebJobs.ServiceBus.EntityType.Topic)] IAsyncCollector<AccountEventDTO> accountCreatedTopic)
{
    log.Info("C# HTTP trigger function processed a request.");

    // Get request body
    var accountEvent = await req.Content.ReadAsAsync<AccountEventDTO>();
    var payload = req.Content.ReadAsStringAsync().Result;

    if (accountEvent != null && accountEvent.Name != null)
    {
        await accountCreatedTopic.AddAsync(accountEvent);
        return req.CreateResponse(HttpStatusCode.OK, "Account successfully added to topic.");
    }

    return req.CreateResponse(HttpStatusCode.BadRequest, "Account was not formed well.");
}

Answer 1

Rather than binding to your AccountEventDTO directly, you'll need to bind to Message (Azure Functions v2) or BrokeredMessage (Azure Functions v1). Then you can set the SessionId property on the message.

To set the body of the message, serialize your DTO as JSON and UTF-8 encode it:

var bytes = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(accountEvent));
var message = new Message(bytes) { SessionId = sessionId };

for v2 or

var bytes = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(accountEvent));
var memoryStream = new MemoryStream(bytes, writable: false);
var message = new BrokeredMessage(memoryStream) { SessionId = sessionId };

for v1.