Class KafkaProducer

Produces to an endpoint.

Inheritance

object

Producer

Producer<KafkaBroker, KafkaProducerEndpoint>

KafkaProducer

KafkaTransactionalProducer

Implements

IProducer

IDisposable

Inherited Members

Producer<KafkaBroker, KafkaProducerEndpoint>.Broker

Producer<KafkaBroker, KafkaProducerEndpoint>.Endpoint

Producer.Id

Producer.Broker

Producer.Endpoint

Producer.IsConnecting

Producer.IsConnected

Producer.ConnectAsync()

Producer.DisconnectAsync()

Producer.Produce(object, IReadOnlyCollection<MessageHeader>)

Producer.Produce(IOutboundEnvelope)

Producer.Produce(object, IReadOnlyCollection<MessageHeader>, Action<IBrokerMessageIdentifier>, Action<Exception>)

Producer.Produce(IOutboundEnvelope, Action<IBrokerMessageIdentifier>, Action<Exception>)

Producer.RawProduce(byte[], IReadOnlyCollection<MessageHeader>)

Producer.RawProduce(Stream, IReadOnlyCollection<MessageHeader>)

Producer.RawProduce(string, byte[], IReadOnlyCollection<MessageHeader>)

Producer.RawProduce(string, Stream, IReadOnlyCollection<MessageHeader>)

Producer.RawProduce(byte[], IReadOnlyCollection<MessageHeader>, Action<IBrokerMessageIdentifier>, Action<Exception>)

Producer.RawProduce(Stream, IReadOnlyCollection<MessageHeader>, Action<IBrokerMessageIdentifier>, Action<Exception>)

Producer.RawProduce(string, byte[], IReadOnlyCollection<MessageHeader>, Action<IBrokerMessageIdentifier>, Action<Exception>)

Producer.RawProduce(string, Stream, IReadOnlyCollection<MessageHeader>, Action<IBrokerMessageIdentifier>, Action<Exception>)

Producer.ProduceAsync(object, IReadOnlyCollection<MessageHeader>)

Producer.ProduceAsync(IOutboundEnvelope)

Producer.ProduceAsync(object, IReadOnlyCollection<MessageHeader>, Action<IBrokerMessageIdentifier>, Action<Exception>)

Producer.ProduceAsync(IOutboundEnvelope, Action<IBrokerMessageIdentifier>, Action<Exception>)

Producer.RawProduceAsync(byte[], IReadOnlyCollection<MessageHeader>)

Producer.RawProduceAsync(Stream, IReadOnlyCollection<MessageHeader>)

Producer.RawProduceAsync(string, byte[], IReadOnlyCollection<MessageHeader>)

Producer.RawProduceAsync(string, Stream, IReadOnlyCollection<MessageHeader>)

Producer.RawProduceAsync(byte[], IReadOnlyCollection<MessageHeader>, Action<IBrokerMessageIdentifier>, Action<Exception>)

Producer.RawProduceAsync(Stream, IReadOnlyCollection<MessageHeader>, Action<IBrokerMessageIdentifier>, Action<Exception>)

Producer.RawProduceAsync(string, byte[], IReadOnlyCollection<MessageHeader>, Action<IBrokerMessageIdentifier>, Action<Exception>)

Producer.RawProduceAsync(string, Stream, IReadOnlyCollection<MessageHeader>, Action<IBrokerMessageIdentifier>, Action<Exception>)

Producer.ConnectCoreAsync()

Producer.DisconnectCoreAsync()

object.Equals(object)

object.Equals(object, object)

object.GetHashCode()

object.GetType()

object.MemberwiseClone()

object.ReferenceEquals(object, object)

object.ToString()

Namespace: Silverback.Messaging.Broker

Assembly: Silverback.Integration.Kafka.dll

Syntax

public class KafkaProducer : Producer<KafkaBroker, KafkaProducerEndpoint>, IProducer, IDisposable

Constructors

KafkaProducer(KafkaBroker, KafkaProducerEndpoint, IBrokerBehaviorsProvider<IProducerBehavior>, IConfluentProducersCache, IServiceProvider, IOutboundLogger<KafkaProducer>)

Initializes a new instance of the KafkaProducer class.

Declaration

public KafkaProducer(KafkaBroker broker, KafkaProducerEndpoint endpoint, IBrokerBehaviorsProvider<IProducerBehavior> behaviorsProvider, IConfluentProducersCache producersCache, IServiceProvider serviceProvider, IOutboundLogger<KafkaProducer> logger)

Parameters

Type	Name	Description
KafkaBroker	broker	The IBroker that instantiated this producer.
KafkaProducerEndpoint	endpoint	The endpoint to produce to.
IBrokerBehaviorsProvider<IProducerBehavior>	behaviorsProvider	The IBrokerBehaviorsProvider<TBehavior>.
IConfluentProducersCache	producersCache	The IConfluentProducersCache.
IServiceProvider	serviceProvider	The IServiceProvider to be used to resolve the required services.
IOutboundLogger<KafkaProducer>	logger	The IOutboundLogger<TCategoryName>.

Methods

Dispose()

Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.

Declaration

public void Dispose()

Dispose(bool)

Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.

Declaration

protected virtual void Dispose(bool disposing)

Parameters

Type	Name	Description
bool	disposing	A value indicating whether the method has been called by the Dispose method and not from the finalizer.

GetConfluentProducer()

Returns the cached Confluent.Kafka.IProducer<TKey, TValue> or creates a new one.

Declaration

protected IProducer<byte[]?, byte[]?> GetConfluentProducer()

Returns

Type	Description
IProducer<byte[], byte[]>	The Confluent.Kafka.IProducer<TKey, TValue>.

ProduceCore(object?, byte[]?, IReadOnlyCollection<MessageHeader>?, string)

Publishes the specified message and returns its identifier.

Declaration

protected override IBrokerMessageIdentifier? ProduceCore(object? message, byte[]? messageBytes, IReadOnlyCollection<MessageHeader>? headers, string actualEndpointName)

Parameters

Type	Name	Description
object	message	The message to be delivered before serialization. This might be null if RawProduce(byte[]?, IReadOnlyCollection<MessageHeader>?), RawProduce(Stream?, IReadOnlyCollection<MessageHeader>?), RawProduceAsync(byte[]?, IReadOnlyCollection<MessageHeader>?) or RawProduceAsync(Stream?, IReadOnlyCollection<MessageHeader>?) have been used to produce.
byte[]	messageBytes	The actual serialized message to be delivered.
IReadOnlyCollection<MessageHeader>	headers	The message headers.
string	actualEndpointName	The actual endpoint to produce to.

Returns

Type	Description
IBrokerMessageIdentifier	The message identifier assigned by the broker (the Kafka offset or similar).

Overrides

Producer.ProduceCore(object?, byte[]?, IReadOnlyCollection<MessageHeader>?, string)

ProduceCore(object?, byte[]?, IReadOnlyCollection<MessageHeader>?, string, Action<IBrokerMessageIdentifier?>, Action<Exception>)

Publishes the specified message and returns its identifier.

Declaration

protected override void ProduceCore(object? message, byte[]? messageBytes, IReadOnlyCollection<MessageHeader>? headers, string actualEndpointName, Action<IBrokerMessageIdentifier?> onSuccess, Action<Exception> onError)

Parameters

Type	Name	Description
object	message	The message to be delivered before serialization. This might be null if RawProduce(byte[]?, IReadOnlyCollection<MessageHeader>?), RawProduce(Stream?, IReadOnlyCollection<MessageHeader>?), RawProduceAsync(byte[]?, IReadOnlyCollection<MessageHeader>?) or RawProduceAsync(Stream?, IReadOnlyCollection<MessageHeader>?) have been used to produce.
byte[]	messageBytes	The actual serialized message to be delivered.
IReadOnlyCollection<MessageHeader>	headers	The message headers.
string	actualEndpointName	The actual endpoint to produce to.
Action<IBrokerMessageIdentifier>	onSuccess	The callback to be invoked when the message is successfully produced.
Action<Exception>	onError	The callback to be invoked when the produce fails.

Overrides

Producer.ProduceCore(object?, byte[]?, IReadOnlyCollection<MessageHeader>?, string, Action<IBrokerMessageIdentifier?>, Action<Exception>)

Remarks

In this implementation the message is synchronously enqueued but produced asynchronously. The callbacks are called when the message is actually produced (or the produce failed).

ProduceCore(object?, Stream?, IReadOnlyCollection<MessageHeader>?, string)

Publishes the specified message and returns its identifier.

Declaration

protected override IBrokerMessageIdentifier? ProduceCore(object? message, Stream? messageStream, IReadOnlyCollection<MessageHeader>? headers, string actualEndpointName)

Parameters

Type	Name	Description
object	message	The message to be delivered before serialization. This might be null if RawProduce(byte[]?, IReadOnlyCollection<MessageHeader>?), RawProduce(Stream?, IReadOnlyCollection<MessageHeader>?), RawProduceAsync(byte[]?, IReadOnlyCollection<MessageHeader>?) or RawProduceAsync(Stream?, IReadOnlyCollection<MessageHeader>?) have been used to produce.
Stream	messageStream	The actual serialized message to be delivered.
IReadOnlyCollection<MessageHeader>	headers	The message headers.
string	actualEndpointName	The actual endpoint to produce to.

Returns

Type	Description
IBrokerMessageIdentifier	The message identifier assigned by the broker (the Kafka offset or similar).

Overrides

Producer.ProduceCore(object?, Stream?, IReadOnlyCollection<MessageHeader>?, string)

ProduceCore(object?, Stream?, IReadOnlyCollection<MessageHeader>?, string, Action<IBrokerMessageIdentifier?>, Action<Exception>)

Publishes the specified message and returns its identifier.

Declaration

protected override void ProduceCore(object? message, Stream? messageStream, IReadOnlyCollection<MessageHeader>? headers, string actualEndpointName, Action<IBrokerMessageIdentifier?> onSuccess, Action<Exception> onError)

Parameters

Type	Name	Description
object	message	The message to be delivered before serialization. This might be null if RawProduce(byte[]?, IReadOnlyCollection<MessageHeader>?), RawProduce(Stream?, IReadOnlyCollection<MessageHeader>?), RawProduceAsync(byte[]?, IReadOnlyCollection<MessageHeader>?) or RawProduceAsync(Stream?, IReadOnlyCollection<MessageHeader>?) have been used to produce.
Stream	messageStream	The message to be delivered.
IReadOnlyCollection<MessageHeader>	headers	The message headers.
string	actualEndpointName	The actual endpoint to produce to.
Action<IBrokerMessageIdentifier>	onSuccess	The callback to be invoked when the message is successfully produced.
Action<Exception>	onError	The callback to be invoked when the produce fails.

Overrides

Producer.ProduceCore(object?, Stream?, IReadOnlyCollection<MessageHeader>?, string, Action<IBrokerMessageIdentifier?>, Action<Exception>)

Remarks

In this implementation the message is synchronously enqueued but produced asynchronously. The callbacks are called when the message is actually produced (or the produce failed).

ProduceCoreAsync(object?, byte[]?, IReadOnlyCollection<MessageHeader>?, string)

Publishes the specified message and returns its identifier.

Declaration

protected override Task<IBrokerMessageIdentifier?> ProduceCoreAsync(object? message, byte[]? messageBytes, IReadOnlyCollection<MessageHeader>? headers, string actualEndpointName)

Parameters

Type	Name	Description
object	message	The message to be delivered before serialization. This might be null if RawProduce(byte[]?, IReadOnlyCollection<MessageHeader>?), RawProduce(Stream?, IReadOnlyCollection<MessageHeader>?), RawProduceAsync(byte[]?, IReadOnlyCollection<MessageHeader>?) or RawProduceAsync(Stream?, IReadOnlyCollection<MessageHeader>?) have been used to produce.
byte[]	messageBytes	The actual serialized message to be delivered.
IReadOnlyCollection<MessageHeader>	headers	The message headers.
string	actualEndpointName	The actual endpoint to produce to.

Returns

Type	Description
Task<IBrokerMessageIdentifier>	A Task<TResult> representing the asynchronous operation. The task result contains the message identifier assigned by the broker (the Kafka offset or similar).

Overrides

Producer.ProduceCoreAsync(object?, byte[]?, IReadOnlyCollection<MessageHeader>?, string)

ProduceCoreAsync(object?, byte[]?, IReadOnlyCollection<MessageHeader>?, string, Action<IBrokerMessageIdentifier?>, Action<Exception>)

Publishes the specified message and returns its identifier.

Declaration

protected override Task ProduceCoreAsync(object? message, byte[]? messageBytes, IReadOnlyCollection<MessageHeader>? headers, string actualEndpointName, Action<IBrokerMessageIdentifier?> onSuccess, Action<Exception> onError)

Parameters

Type	Name	Description
object	message	The message to be delivered before serialization. This might be null if RawProduce(byte[]?, IReadOnlyCollection<MessageHeader>?), RawProduce(Stream?, IReadOnlyCollection<MessageHeader>?), RawProduceAsync(byte[]?, IReadOnlyCollection<MessageHeader>?) or RawProduceAsync(Stream?, IReadOnlyCollection<MessageHeader>?) have been used to produce.
byte[]	messageBytes	The actual serialized message to be delivered.
IReadOnlyCollection<MessageHeader>	headers	The message headers.
string	actualEndpointName	The actual endpoint to produce to.
Action<IBrokerMessageIdentifier>	onSuccess	The callback to be invoked when the message is successfully produced.
Action<Exception>	onError	The callback to be invoked when the produce fails.

Returns

Type	Description
Task	A Task representing the asynchronous operation. The Task will complete as soon as the message is enqueued.

Overrides

Producer.ProduceCoreAsync(object?, byte[]?, IReadOnlyCollection<MessageHeader>?, string, Action<IBrokerMessageIdentifier?>, Action<Exception>)

Remarks

The returned Task completes when the message is enqueued while the callbacks are called when the message is actually produced (or the produce failed).

ProduceCoreAsync(object?, Stream?, IReadOnlyCollection<MessageHeader>?, string)

Publishes the specified message and returns its identifier.

Declaration

protected override Task<IBrokerMessageIdentifier?> ProduceCoreAsync(object? message, Stream? messageStream, IReadOnlyCollection<MessageHeader>? headers, string actualEndpointName)

Parameters

Type	Name	Description
object	message	The message to be delivered before serialization. This might be null if RawProduce(byte[]?, IReadOnlyCollection<MessageHeader>?), RawProduce(Stream?, IReadOnlyCollection<MessageHeader>?), RawProduceAsync(byte[]?, IReadOnlyCollection<MessageHeader>?) or RawProduceAsync(Stream?, IReadOnlyCollection<MessageHeader>?) have been used to produce.
Stream	messageStream	The message to be delivered.
IReadOnlyCollection<MessageHeader>	headers	The message headers.
string	actualEndpointName	The actual endpoint to produce to.

Returns

Type	Description
Task<IBrokerMessageIdentifier>	A Task<TResult> representing the asynchronous operation. The task result contains the message identifier assigned by the broker (the Kafka offset or similar).

Overrides

Producer.ProduceCoreAsync(object?, Stream?, IReadOnlyCollection<MessageHeader>?, string)

ProduceCoreAsync(object?, Stream?, IReadOnlyCollection<MessageHeader>?, string, Action<IBrokerMessageIdentifier?>, Action<Exception>)

Publishes the specified message and returns its identifier.

Declaration

protected override Task ProduceCoreAsync(object? message, Stream? messageStream, IReadOnlyCollection<MessageHeader>? headers, string actualEndpointName, Action<IBrokerMessageIdentifier?> onSuccess, Action<Exception> onError)

Parameters

Type	Name	Description
object	message	The message to be delivered before serialization. This might be null if RawProduce(byte[]?, IReadOnlyCollection<MessageHeader>?), RawProduce(Stream?, IReadOnlyCollection<MessageHeader>?), RawProduceAsync(byte[]?, IReadOnlyCollection<MessageHeader>?) or RawProduceAsync(Stream?, IReadOnlyCollection<MessageHeader>?) have been used to produce.
Stream	messageStream	The message to be delivered.
IReadOnlyCollection<MessageHeader>	headers	The message headers.
string	actualEndpointName	The actual endpoint to produce to.
Action<IBrokerMessageIdentifier>	onSuccess	The callback to be invoked when the message is successfully produced.
Action<Exception>	onError	The callback to be invoked when the produce fails.

Returns

Type	Description
Task	A Task representing the asynchronous operation. The Task will complete as soon as the message is enqueued.

Overrides

Producer.ProduceCoreAsync(object?, Stream?, IReadOnlyCollection<MessageHeader>?, string, Action<IBrokerMessageIdentifier?>, Action<Exception>)

Remarks

The returned Task completes when the message is enqueued while the callbacks are called when the message is actually produced (or the produce failed).

Implements

IProducer

IDisposable