dbarkol

public class KafkaAvroAsyncSerializer<T> : IAsyncSerializer<T>
{
    private readonly SchemaRegistryAvroObjectSerializer serializer;

    public KafkaAvroAsyncSerializer(string schemaRegistryUrl, TokenCredential credential, string schemaGroup, Boolean autoRegisterSchemas = false)
    {
        this.serializer = new SchemaRegistryAvroObjectSerializer(
            new SchemaRegistryClient(
                schemaRegistryUrl,
                credential),

dbarkol

public T Deserialize(ReadOnlySpan<byte> data, bool isNull, SerializationContext context)
{
    if (data.IsEmpty)
    {
        return default(T);
    }

    return (T) this.serializer.Deserialize(new MemoryStream(data.ToArray()), typeof(T), CancellationToken.None);
}

dbarkol

using Azure.Identity;
using Confluent.Kafka;
using Microsoft.Azure.Kafka.SchemaRegistry.Avro;
using System;
using System.Configuration;
using System.Threading.Tasks;
using zohan.schemaregistry.events;

namespace Zohan.SchemaRegistry.Producer
{

dbarkol

using Azure.Identity;
using Confluent.Kafka;
using Microsoft.Azure.Kafka.SchemaRegistry.Avro;
using System;
using System.Configuration;
using System.Threading;
using zohan.schemaregistry.events;

namespace Zohan.SchemaRegistry.Consumer
{

dbarkol

{
    "specversion" : "1.0",
    "id" : "b85d631a-101e-005a-02f2-cee7aa06f148",
    "type" : "zohan.music.request",
    "source" : "https://zohan.dev/music/",
    "subject" : "zohan/music/requests/4322",
    "time" : "2020-09-14T10:00:00Z",
    "data" : {
      "artist": "Gerardo",
      "song": "Rico Suave"

dbarkol

{
    "specversion" : "1.0",
    "id" : "b85d631a-101e-005a-02f2-cee7aa06f148",
    "type" : "contoso.loyalty.update",
    "source" : "https://contoso.com/loyalty",
    "subject" : "contoso/loyalty/points",
    "time" : "2021-06-14T10:00:00Z",
    "data" : {      
      "loyaltyId": "100", 
      "points": "400",

dbarkol

using Microsoft.Azure.Cosmos;
using Microsoft.Azure.Functions.Extensions.DependencyInjection;
using Microsoft.Extensions.DependencyInjection;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

[assembly: FunctionsStartup(typeof(OrderMaker.Startup))]

dbarkol

using System;
using System.IO;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Extensions.Http;
using Microsoft.AspNetCore.Http;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
using OrderMaker.Models;

dbarkol

  [FunctionName("OrderOutboxWorker")]
  public static async Task Run(
      [CosmosDBTrigger(
          databaseName: "%CosmosDBDatabaseName%",
          collectionName: "%CosmosDBCollectionName%",
          ConnectionStringSetting = "CosmosDBConnectionString",
          CreateLeaseCollectionIfNotExists = true,
          LeaseCollectionName = "%CosmosDBLeaseCollectionName%")] IReadOnlyList<Document> input,
      [CosmosDB(
          databaseName: "%CosmosDBDatabaseName%",

dbarkol

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Azure.Messaging.ServiceBus;
using Microsoft.Azure.Cosmos;
using Microsoft.Azure.Documents;
using Microsoft.Azure.Documents.Client;
using Microsoft.Azure.Documents.Linq;