Sandbox - Docs - Twitter - YouTube
Created and maintained by the Memphis community.
Maintainer: @bazen-teklehaymanot
Memphis is a next-generation alternative to traditional message brokers.
A simple, robust, and durable cloud-native message broker wrapped with
an entire ecosystem that enables cost-effective, fast, and reliable development of modern queue-based use cases.
Memphis enables the building of modern queue-based applications that require
large volumes of streamed and enriched data, modern protocols, zero ops, rapid development,
extreme cost reduction, and a significantly lower amount of dev time for data-oriented developers and data engineers.
dotnet add package Memphis.Client -v ${MEMPHIS_CLIENT_VERSION}
Update-Package Memphis.Client
using Memphis.Client;
First, we need to create or use default ClientOptions
and then connect with Memphis by using MemphisClientFactory.CreateClient(ClientOptions opst)
.
try
{
var options = MemphisClientFactory.GetDefaultOptions();
options.Host = "<broker-address>";
options.Username = "<application-type-username>";
options.ConnectionToken = "<token>"; // you will get it on application type user creation
var memphisClient = MemphisClientFactory.CreateClient(options);
...
}
catch (Exception ex)
{
Console.Error.WriteLine("Exception: " + ex.Message);
Console.Error.WriteLine(ex);
}
Once client created, the entire functionalities offered by Memphis are available.
To disconnect from Memphis, call Dispose()
on the MemphisClient
.
await memphisClient.Dispose()
try
{
// First: creating Memphis client
var options = MemphisClientFactory.GetDefaultOptions();
options.Host = "<memphis-host>";
options.Username = "<application type username>";
options.ConnectionToken = "<broker-token>";
var client = MemphisClientFactory.CreateClient(options);
// Second: creaing Memphis station
var station = await client.CreateStation(
stationOptions: new StationOptions()
{
Name = "<station-name>",
RetentionType = RetentionTypes.MAX_MESSAGE_AGE_SECONDS,
RetentionValue = 604_800,
StorageType = StorageTypes.DISK,
Replicas = 1,
IdempotencyWindowMs = 0,
SendPoisonMessageToDls = true,
SendSchemaFailedMessageToDls = true,
});
}
catch (Exception ex)
{
Console.Error.WriteLine("Exception: " + ex.Message);
Console.Error.WriteLine(ex);
}
Memphis currently supports the following types of retention:
RetentionTypes.MAX_MESSAGE_AGE_SECONDS
The above means that every message persists for the value set in the retention value field (in seconds).
RetentionTypes.MESSAGES
The above means that after the maximum number of saved messages (set in retention value) has been reached, the oldest messages will be deleted.
RetentionTypes.BYTES
The above means that after maximum number of saved bytes (set in retention value) has been reached, the oldest messages will be deleted.
The retention values
are directly related to the retention types
mentioned above,
where the values vary according to the type of retention chosen.
All retention values are of type int
but with different representations as follows:
RetentionTypes.MAX_MESSAGE_AGE_SECONDS
is represented in seconds, RetentionTypes.MESSAGES
in a number of messages
and finally RetentionTypes.BYTES
in a number of bytes.
After these limits are reached oldest messages will be deleted.
Memphis currently supports the following types of messages storage:
StorageTypes.DISK
The above means that messages persist on disk.
StorageTypes.MEMORY
The above means that messages persist on the main memory.
Destroying a station will remove all its resources (including producers and consumers).
station.DestroyAsync()
await client.AttachSchema(stationName: "<station-name>", schemaName: "<schema-name>");
await client.DetachSchema(stationName: station.Name);
The most common client operations are produce
to send messages and consume
to
receive messages.
Messages are published to a station and consumed from it by creating a consumer. Consumers are pull based and consume all the messages in a station unless you are using a consumers group, in this case messages are spread across all members in this group.
Memphis messages are payload agnostic. Payloads are byte[]
.
In order to stop getting messages, you have to call consumer.Dispose()
. Destroy will terminate regardless
of whether there are messages in flight for the client.
try
{
// First: creating Memphis client
var options = MemphisClientFactory.GetDefaultOptions();
options.Host = "<memphis-host>";
options.Username = "<application type username>";
options.ConnectionToken = "<broker-token>";
var client = MemphisClientFactory.CreateClient(options);
// Second: creating the Memphis producer
var producer = await client.CreateProducer(
stationName: "<memphis-station-name>",
producerName: "<memphis-producer-name>",
generateRandomSuffix:true);
}
catch (Exception ex)
{
Console.Error.WriteLine("Exception: " + ex.Message);
Console.Error.WriteLine(ex);
}
var commonHeaders = new NameValueCollection();
commonHeaders.Add("key-1", "value-1");
await producer.ProduceAsync(Encoding.UTF8.GetBytes(text), commonHeaders);
await producer.DestroyAsync()
try
{
// First: creating Memphis client
var options = MemphisClientFactory.GetDefaultOptions();
options.Host = "<memphis-host>";
options.Username = "<application type username>";
options.ConnectionToken = "<broker-token>";
var client = MemphisClientFactory.CreateClient(options);
// Second: creaing Memphis consumer
var consumer = await client.CreateConsumer(new ConsumerOptions
{
StationName = "<station-name>",
ConsumerName = "<consumer-name>",
ConsumerGroup = "<consumer-group-name>",
});
}
catch (Exception ex)
{
Console.Error.WriteLine("Exception: " + ex.Message);
Console.Error.WriteLine(ex);
}
First, create a callback functions that receives a args that holds list of MemhpisMessage. Then, pass this callback into consumer.Consume function. The consumer will try to fetch messages every PullIntervalMs (that was given in Consumer's creation) and call the defined message handler.
EventHandler<MemphisMessageHandlerEventArgs> msgHandler = (sender, args) =>
{
if (args.Exception != null)
{
Console.Error.WriteLine(args.Exception);
return;
}
foreach (var msg in args.MessageList)
{
//print message itself
Console.WriteLine("Received data: " + Encoding.UTF8.GetString(msg.GetData()));
// print message headers
foreach (var headerKey in msg.GetHeaders().Keys)
{
Console.WriteLine(
$"Header Key: {headerKey}, value: {msg.GetHeaders()[headerKey.ToString()]}");
}
Console.WriteLine("---------");
msg.Ack();
}
};
await consumer.ConsumeAsync( msgCallbackHandler:msgHandler, dlqCallbackHandler:msgHandler);
Acknowledging a message indicates to the Memphis server to not re-send the same message again to the same consumer or consumers group.
msg.Ack();
Get headers per message
msg.GetHeaders()
await consumer.DestroyAsync();
memphisClient.IsConnected();