@ -17,267 +17,266 @@ using Squidex.Infrastructure.Json;
using Squidex.Infrastructure.Json.Objects ; using Squidex.Infrastructure.Json.Objects ;
using Schema = Avro . Schema ; using Schema = Avro . Schema ;
namespace Squidex.Extensions.Actions.Kafka namespace Squidex.Extensions.Actions.Kafka ;
{
public sealed class KafkaProducer
{
private readonly IProducer < string , string > textProducer ;
private readonly IProducer < string , GenericRecord > avroProducer ;
private readonly ISchemaRegistryClient schemaRegistry ;
private readonly IJsonSerializer jsonSerializer ;
public KafkaProducer ( IOptions < KafkaProducerOptions > options , IJsonSerializer jsonSerializer , public sealed class KafkaProducer
ILogger < KafkaProducer > log ) {
{ private readonly IProducer < string , string > textProducer ;
this . jsonSerializer = jsonSerializer ; private readonly IProducer < string , GenericRecord > avroProducer ;
private readonly ISchemaRegistryClient schemaRegistry ;
private readonly IJsonSerializer jsonSerializer ;
textProducer = new ProducerBuilder < string , string > ( options . Value ) public KafkaProducer ( IOptions < KafkaProducerOptions > options , IJsonSerializer jsonSerializer ,
. SetErrorHandler ( ( p , error ) = > ILogger < KafkaProducer > log )
{ {
LogError ( log , error ) ; this . jsonSerializer = jsonSerializer ;
} )
. SetLogHandler ( ( p , message ) = >
{
LogMessage ( log , message ) ;
} )
. SetKeySerializer ( Serializers . Utf8 )
. SetValueSerializer ( Serializers . Utf8 )
. Build ( ) ;
if ( options . Value . IsSchemaRegistryConfigured ( ) ) textProducer = new ProducerBuilder < string , string > ( options . Value )
. SetErrorHandler ( ( p , error ) = >
{ {
schemaRegistry = new CachedSchemaRegistryClient ( options . Value . SchemaRegistry ) ; LogError ( log , error ) ;
} )
avroProducer = new ProducerBuilder < string , GenericRecord > ( options . Value ) . SetLogHandler ( ( p , message ) = >
. SetErrorHandler ( ( p , error ) = >
{
LogError ( log , error ) ;
} )
. SetLogHandler ( ( p , message ) = >
{
LogMessage ( log , message ) ;
} )
. SetKeySerializer ( Serializers . Utf8 )
. SetValueSerializer ( new AvroSerializer < GenericRecord > ( schemaRegistry , options . Value . AvroSerializer ) )
. Build ( ) ;
}
}
private static void LogMessage ( ILogger < KafkaProducer > log , LogMessage message )
{
var level = LogLevel . Information ;
switch ( message . Level )
{ {
case SyslogLevel . Emergency : LogMessage ( log , message ) ;
level = LogLevel . Error ; } )
break ; . SetKeySerializer ( Serializers . Utf8 )
case SyslogLevel . Alert : . SetValueSerializer ( Serializers . Utf8 )
level = LogLevel . Error ; . Build ( ) ;
break ;
case SyslogLevel . Critical :
level = LogLevel . Error ;
break ;
case SyslogLevel . Error :
level = LogLevel . Error ;
break ;
case SyslogLevel . Warning :
level = LogLevel . Warning ;
break ;
case SyslogLevel . Notice :
level = LogLevel . Information ;
break ;
case SyslogLevel . Info :
level = LogLevel . Information ;
break ;
case SyslogLevel . Debug :
level = LogLevel . Debug ;
break ;
}
log . Log ( level , "Kafka log {name}: {message}." , message . Name , message . Message ) ; if ( options . Value . IsSchemaRegistryConfigured ( ) )
{
schemaRegistry = new CachedSchemaRegistryClient ( options . Value . SchemaRegistry ) ;
avroProducer = new ProducerBuilder < string , GenericRecord > ( options . Value )
. SetErrorHandler ( ( p , error ) = >
{
LogError ( log , error ) ;
} )
. SetLogHandler ( ( p , message ) = >
{
LogMessage ( log , message ) ;
} )
. SetKeySerializer ( Serializers . Utf8 )
. SetValueSerializer ( new AvroSerializer < GenericRecord > ( schemaRegistry , options . Value . AvroSerializer ) )
. Build ( ) ;
} }
}
private static void LogMessage ( ILogger < KafkaProducer > log , LogMessage message )
{
var level = LogLevel . Information ;
private static void LogError ( ILogger < KafkaProducer > log , Error error ) switch ( message . Level )
{ {
log . LogWarning ( "Kafka error with {code} and {reason}." , error . Code , error . Reason ) ; case SyslogLevel . Emergency :
level = LogLevel . Error ;
break ;
case SyslogLevel . Alert :
level = LogLevel . Error ;
break ;
case SyslogLevel . Critical :
level = LogLevel . Error ;
break ;
case SyslogLevel . Error :
level = LogLevel . Error ;
break ;
case SyslogLevel . Warning :
level = LogLevel . Warning ;
break ;
case SyslogLevel . Notice :
level = LogLevel . Information ;
break ;
case SyslogLevel . Info :
level = LogLevel . Information ;
break ;
case SyslogLevel . Debug :
level = LogLevel . Debug ;
break ;
} }
public async Task SendAsync ( KafkaJob job , log . Log ( level , "Kafka log {name}: {message}." , message . Name , message . Message ) ;
CancellationToken ct ) }
private static void LogError ( ILogger < KafkaProducer > log , Error error )
{
log . LogWarning ( "Kafka error with {code} and {reason}." , error . Code , error . Reason ) ;
}
public async Task SendAsync ( KafkaJob job ,
CancellationToken ct )
{
if ( ! string . IsNullOrWhiteSpace ( job . Schema ) )
{ {
if ( ! string . IsNullOrWhiteSpace ( job . Schema ) ) var value = CreateAvroRecord ( job . MessageValue , job . Schema ) ;
{
var value = CreateAvroRecord ( job . MessageValue , job . Schema ) ;
var message = new Message < string , GenericRecord > { Value = value } ; var message = new Message < string , GenericRecord > { Value = value } ;
await ProduceAsync ( avroProducer , message , job , ct ) ; await ProduceAsync ( avroProducer , message , job , ct ) ;
} }
else else
{ {
var message = new Message < string , string > { Value = job . MessageValue } ; var message = new Message < string , string > { Value = job . MessageValue } ;
await ProduceAsync ( textProducer , message , job , ct ) ; await ProduceAsync ( textProducer , message , job , ct ) ;
}
} }
}
private static async Task ProduceAsync < T > ( IProducer < string , T > producer , Message < string , T > message , KafkaJob job ,
CancellationToken ct )
{
message . Key = job . MessageKey ;
private static async Task ProduceAsync < T > ( IProducer < string , T > producer , Message < string , T > message , KafkaJob job , if ( job . Headers ? . Count > 0 )
CancellationToken ct )
{ {
message . Key = job . MessageKey ; message . Headers = new Headers ( ) ;
if ( job . Headers ? . Count > 0 ) foreach ( var header in job . Headers )
{ {
message . Headers = new Headers ( ) ; message . Headers . Add ( header . Key , Encoding . UTF8 . GetBytes ( header . Value ) ) ;
foreach ( var header in job . Headers )
{
message . Headers . Add ( header . Key , Encoding . UTF8 . GetBytes ( header . Value ) ) ;
}
} }
}
if ( ! string . IsNullOrWhiteSpace ( job . PartitionKey ) & & job . PartitionCount > 0 ) if ( ! string . IsNullOrWhiteSpace ( job . PartitionKey ) & & job . PartitionCount > 0 )
{ {
var partition = Math . Abs ( job . PartitionKey . GetHashCode ( StringComparison . Ordinal ) ) % job . PartitionCount ; var partition = Math . Abs ( job . PartitionKey . GetHashCode ( StringComparison . Ordinal ) ) % job . PartitionCount ;
await producer . ProduceAsync ( new TopicPartition ( job . TopicName , partition ) , message , ct ) ; await producer . ProduceAsync ( new TopicPartition ( job . TopicName , partition ) , message , ct ) ;
} }
else else
{ {
await producer . ProduceAsync ( job . TopicName , message , ct ) ; await producer . ProduceAsync ( job . TopicName , message , ct ) ;
}
} }
}
private GenericRecord CreateAvroRecord ( string json , string avroSchema ) private GenericRecord CreateAvroRecord ( string json , string avroSchema )
{
try
{ {
try var schema = ( RecordSchema ) Schema . Parse ( avroSchema ) ;
{
var schema = ( RecordSchema ) Schema . Parse ( avroSchema ) ;
var jsonObject = jsonSerializer . Deserialize < JsonObject > ( json ) ; var jsonObject = jsonSerializer . Deserialize < JsonObject > ( json ) ;
var result = ( GenericRecord ) GetValue ( jsonObject , schema ) ; var result = ( GenericRecord ) GetValue ( jsonObject , schema ) ;
return result ; return result ;
}
catch ( JsonException ex )
{
throw new InvalidOperationException ( $"Failed to parse json: {json}, got {ex.Message}" , ex ) ;
}
} }
catch ( JsonException ex )
public void Dispose ( )
{ {
textProducer ? . Dispose ( ) ; throw new InvalidOperationException ( $"Failed to parse json: {json}, got {ex.Message}" , ex ) ;
avroProducer ? . Dispose ( ) ;
} }
}
public void Dispose ( )
{
textProducer ? . Dispose ( ) ;
avroProducer ? . Dispose ( ) ;
}
private static object GetValue ( JsonValue value , Schema schema ) private static object GetValue ( JsonValue value , Schema schema )
{
switch ( value . Value )
{ {
switch ( value . Value ) case bool b when IsTypeOrUnionWith ( schema , Schema . Type . Boolean ) :
{ return b ;
case bool b when IsTypeOrUnionWith ( schema , Schema . Type . Boolean ) : case double d when IsTypeOrUnionWith ( schema , Schema . Type . Long ) :
return b ; return ( long ) d ;
case double d when IsTypeOrUnionWith ( schema , Schema . Type . Long ) : case double d when IsTypeOrUnionWith ( schema , Schema . Type . Float ) :
return ( long ) d ; return ( float ) d ;
case double d when IsTypeOrUnionWith ( schema , Schema . Type . Float ) : case double d when IsTypeOrUnionWith ( schema , Schema . Type . Int ) :
return ( float ) d ; return ( int ) d ;
case double d when IsTypeOrUnionWith ( schema , Schema . Type . Int ) : case double d when IsTypeOrUnionWith ( schema , Schema . Type . Double ) :
return ( int ) d ; return d ;
case double d when IsTypeOrUnionWith ( schema , Schema . Type . Double ) : case string s when IsTypeOrUnionWith ( schema , Schema . Type . String ) :
return d ; return s ;
case string s when IsTypeOrUnionWith ( schema , Schema . Type . String ) : case JsonObject o when IsTypeOrUnionWith ( schema , Schema . Type . Map ) :
return s ; {
case JsonObject o when IsTypeOrUnionWith ( schema , Schema . Type . Map ) : var mapResult = new Dictionary < string , object > ( ) ;
if ( schema is UnionSchema union )
{ {
var mapResult = new Dictionary < string , object > ( ) ; var map = ( MapSchema ) union . Schemas . FirstOrDefault ( x = > x . Tag = = Schema . Type . Map ) ;
if ( schema is UnionSchema union ) foreach ( var ( key , childValue ) in o )
{ {
var map = ( MapSchema ) union . Schemas . FirstOrDefault ( x = > x . Tag = = Schema . Type . Map ) ; mapResult . Add ( key , GetValue ( childValue , map ? . ValueSchema ) ) ;
foreach ( var ( key , childValue ) in o )
{
mapResult . Add ( key , GetValue ( childValue , map ? . ValueSchema ) ) ;
}
} }
else if ( schema is MapSchema map ) }
else if ( schema is MapSchema map )
{
foreach ( var ( key , childValue ) in o )
{ {
foreach ( var ( key , childValue ) in o ) mapResult . Add ( key , GetValue ( childValue , map ? . ValueSchema ) ) ;
{
mapResult . Add ( key , GetValue ( childValue , map ? . ValueSchema ) ) ;
}
} }
return mapResult ;
} }
case JsonObject o when IsTypeOrUnionWith ( schema , Schema . Type . Record ) : return mapResult ;
{ }
GenericRecord result = null ;
if ( schema is UnionSchema union ) case JsonObject o when IsTypeOrUnionWith ( schema , Schema . Type . Record ) :
{ {
var record = ( RecordSchema ) union . Schemas . FirstOrDefault ( x = > x . Tag = = Schema . Type . Record ) ; GenericRecord result = null ;
result = new GenericRecord ( record ) ; if ( schema is UnionSchema union )
{
var record = ( RecordSchema ) union . Schemas . FirstOrDefault ( x = > x . Tag = = Schema . Type . Record ) ;
result = new GenericRecord ( record ) ;
foreach ( var ( key , childValue ) in o ) foreach ( var ( key , childValue ) in o )
{
if ( record ! = null & & record . TryGetField ( key , out var field ) )
{ {
if ( record ! = null & & record . TryGetField ( key , out var field ) ) result . Add ( key , GetValue ( childValue , field . Schema ) ) ;
{
result . Add ( key , GetValue ( childValue , field . Schema ) ) ;
}
} }
} }
else if ( schema is RecordSchema record ) }
{ else if ( schema is RecordSchema record )
result = new GenericRecord ( record ) ; {
result = new GenericRecord ( record ) ;
foreach ( var ( key , childValue ) in o ) foreach ( var ( key , childValue ) in o )
{
if ( record . TryGetField ( key , out var field ) )
{ {
if ( record . TryGetField ( key , out var field ) ) result . Add ( key , GetValue ( childValue , field . Schema ) ) ;
{
result . Add ( key , GetValue ( childValue , field . Schema ) ) ;
}
} }
} }
return result ;
} }
case JsonArray a when IsTypeOrUnionWith ( schema , Schema . Type . Array ) : return result ;
}
case JsonArray a when IsTypeOrUnionWith ( schema , Schema . Type . Array ) :
{
var result = new List < object > ( ) ;
if ( schema is UnionSchema union )
{ {
var result = new List < object > ( ) ; var arraySchema = ( ArraySchema ) union . Schemas . FirstOrDefault ( x = > x . Tag = = Schema . Type . Array ) ;
if ( schema is UnionSchema union ) foreach ( var item in a )
{ {
var arraySchema = ( ArraySchema ) union . Schemas . FirstOrDefault ( x = > x . Tag = = Schema . Type . Array ) ; result . Add ( GetValue ( item , arraySchema ? . ItemSchema ) ) ;
foreach ( var item in a )
{
result . Add ( GetValue ( item , arraySchema ? . ItemSchema ) ) ;
}
} }
else if ( schema is ArraySchema array ) }
else if ( schema is ArraySchema array )
{
foreach ( var item in a )
{ {
foreach ( var item in a ) result . Add ( GetValue ( item , array . ItemSchema ) ) ;
{
result . Add ( GetValue ( item , array . ItemSchema ) ) ;
}
} }
return result . ToArray ( ) ;
} }
}
return null ; return result . ToArray ( ) ;
}
} }
private static bool IsTypeOrUnionWith ( Schema schema , Schema . Type expected ) return null ;
{ }
return schema . Tag = = expected | | ( schema is UnionSchema union & & union . Schemas . Any ( x = > x . Tag = = expected ) ) ;
} private static bool IsTypeOrUnionWith ( Schema schema , Schema . Type expected )
{
return schema . Tag = = expected | | ( schema is UnionSchema union & & union . Schemas . Any ( x = > x . Tag = = expected ) ) ;
} }
} }
Sebastian Stehle
4 years ago
GitHub