arke.proto

/**
 * Arke message broker proxy messages.
 *
 * This file outlines the gRPC interface for the Arke proxy.
 */
syntax = "proto3";

option java_outer_classname = "ArkeProtos";
option java_package = "org.arke";
option go_package = "github.com/sassoftware/arke/api";

import "google/protobuf/timestamp.proto";

package arke;

/**
 * Represents the broker connection information. This is passed
 * in by the client allowing us to remain a proxy, but also
 * allowing us to support multiple providers at once such as
 * RabbitMQ and Kafka.
 */
message ConnectionConfiguration {
    string host = 1; // Broker hostname or IP address.
    int32 port = 2; // Broker port.
    string provider = 3; // Provider type, currently only amqp091.
    string tenant = 4; // Tenant name for this connection. Tenant is not required
    Credentials credentials = 5; // Authentication credentials.
    bool tls = 8; // Should this provider connection use TLS.
    string client_name = 9; // The name of the client connecting.
    int32 admin_port = 10; // The administrative port for the provider (eg. RabbitMQ management port) for any actions needing to be performed by the provider (eg. modifying bindings for RabbitMQ)

    reserved 6, 7, 11;
    reserved "prefetch_count", "ca_certificate", "publisher_name";
}

/**
 * Represents the broker authentication information.
 */
message Credentials {
    string username = 1; // Username for authenticating to broker.
    string password = 2; // Password for authenticating to broker.
}

/**
 * Represents the connection response
 */
message ConnectResponse {
    bool success = 1; // Indicates whether the connection was successful.
    Error error = 2; // Error if the connection failed.
}

/**
 * Represents a message that is being produced or consumed. The
 * error property is set by the proxy when an error occurs during
 * the consumer process.
 */
message Message {
    string uuid = 1; // The proxy sets this UUID when the message is consumed. It should be returned when a message is Ack/Nack'ed.
    map<string,string> headers = 2; // The message headers.
    bytes body = 3; // The message body.
    Address address = 4; // The distination for a published message.
    bool persistent = 5; // Indicates whether to persist the message.
    Error error = 6; // Error message if consuming failed.
    bool confirm = 7; // Enables guaranteed delivery to the broker.
    int64 publish_id = 8; // Combined with the publisher_name to provide publishing deduplication. The publish_id is a strictly increasing sequence.
    string publisher_name = 11; // The publisher_name is used in conjunction with publish_id to provide message deduplication on Streams ONLY.
}

/**
 * Represents the response from publishing a message.
 */
message MessageResponse {
    bool success = 1; // Indicates whether the message was published successfully.
    Error error = 2; // Error if publishing failed.
}

/**
 * Represents the publishing destination for a message.
 */
message Address {
    string name = 1; // The name of this destination address.
    repeated string subjects = 2; // The subjects of the address. Multiple subjects are allowed on Subscribe, but not on Publish.
    enum TargetType {
        TOPIC = 0; // The address is a message topic.
        QUEUE = 1; // The address is a message queue.
        FILTER = 2; // The address is a filtered queue.
        STREAM = 3; // The address is a stream.
    }
    TargetType type = 3; // Target type, default is TOPIC.
    bool durable = 4 [deprecated = true];
    bool auto_delete = 5; // Should the address automatically delete.
    Address parent_address = 6; // A parent Address. Usage includes Address to Address binding.
}

/**
 * Represents the source for consumer subscriptions. The Stream TargetType
 * does not support exclusive or auto_delete. Streams also do not support the
 * following Options: Exclusive, DeadLetterAdress or DeadLetterSubject.
 */
message Source {
    string name = 1; // The name of this source.
    Address address = 3; // The Address associated with this source.
    bool durable = 4 [deprecated = true];
    bool auto_delete = 5; // Should this Source automatically delete.
    map<string,string> options = 7; // Additional options for this Source. Option keys include: MessageTTL, Expires, DeadLetterAddress, DeadLetterSubject, Offset(Valid values: first, continue, next, or a quoted integer), ConsumerGroup.
    bool exclusive = 8; // Should this source be exclusive to the subscriber.
    int32 prefetch_count = 9; // Set the prefetch count for this subscriber. Must be greater than 0. Defaults to 1.
    repeated Filter filters = 10; // Filters for this Source.
    enum TargetType {
        QUEUE = 0; // The Source is a HA queue
        TEMPORARY = 1; // The Source is a temporary queue
        STREAM = 2; // The Source is a stream
    }
    TargetType type = 11; // Target type, default is QUEUE.
    bool declare_only = 12; // Declare the source and any bindings but do not actually consume any messages.
    bool single_active_consumer = 13; // Only one consumer should be allowed to subscribe to this source. For queues, this applies to all consumers on the queue. For streams this option must be paired with the ConsumerGroup option.

    reserved 6;
    reserved "filter";
}

/**
 * Represents a filter for Message.headers. The Filter will prevent a consumer
 * from consuming all messages from a Source.
 */
message Filter {
    repeated Match matches = 1; // One or more filter Matches.
    enum MatchType {
        ALL = 0; // All matches must match for a successful filter.
        ANY = 1; // Any match can match for a successful filter.
    }
    MatchType type = 2; // The MatchType for this filter. Default is ALL.
}

/**
 * Represents the key/value match for Message.headers. Currently only an
 * exact match is supported.
 */
message Match {
    string name = 1; // The Message.headers key
    string value = 2; // The Message.headers value
}

/**
 * No parameters or return value.
 */
message Empty {}

/**
 * Represents a generic error message.
 */
message Error {
    string message = 1; // The text error message.
    int32 code = 2; // The error code.
    bool is_fatal = 3; // Indicator that a fatal error has occured, and a reconnect is required on this connection.
}

/**
 * Message used to ack or nack a message UUID
 */
message MessageConsumed {
    string uuid = 1; // arke.Message UUID returned from Consume.
    bool nack = 2; // Nack (true) or Ack (false) the message. By default all messages will be Ack'd unless you set this to true.
    int32 requeue_delay = 3; // Requeue delay in seconds for Nack messages. Delay of zero will result in Nack'd messages getting dequeued. If delay is greater than zero, the message will be requeued after the delay.
}

/**
 * Response to a MessageConsumed message
 */
message MessageConsumedResponse {
    string uuid = 1; // Message UUID of the arke.Message
    bool success = 2; // Was the ack or nack successful.
    Error error = 3; // Error if the MessageConsumed failed
}

/**
 * Sent on the Consume stream from the client to either start consuming messages or to ack/nack messages
 */
message Consume {
    oneof msg {
        arke.Source src = 1;
        arke.MessageConsumed ack = 2;
    }
}

/**
 * Response to a Consume request where Source.DeclareOnly is true.
 */
message DeclareOnlyResponse {
    bool success = 1; // Was the source declaration successful.
    Error error = 2; // Error if the source declaration failed.
}

/**
 * Response to a Consume message. Either a Message or a MessageConsumedResponse
 */
message ConsumeResponse {
    oneof resp {
        arke.Message msg = 1;
        arke.MessageConsumedResponse consumed_response = 2;
        arke.Error error = 3;
        arke.DeclareOnlyResponse declare_only_response = 4;
    }
}

/**
 * SourceStats includes information about the source from the broker
 */
 message SourceStats {
    arke.Error error = 1; // Any error encountered while retrieving the SourceStats.
    int64 message_count = 2; // Total number of messages in the queue.
    int32 consumer_count = 3; // Number of consumers on the queue or stream.
    int64 last_offset = 4; // Offset of the last message in the stream.
    int64 current_offset = 5; // Current stream offset for the specified consumer.
   string name = 6; // Name of the source.
   float publish_rate = 7; // The rate of messages being published to this source in messages/second.
   float deliver_rate = 8; // The rate of messages being delivered from this source in messages/second.
}

/**
 * Represents a group of Sources
 */
message Sources {
  repeated arke.Source sources = 1; // The sources we want metrics returned for.
}

/**
 * Return a group of SourceStats for the specified Sources.
 */
message SourceStatsCollection {
  arke.Error error = 1; // Any error encountered while retrieving the SourcesMetrics
  repeated arke.SourceStats stats = 2; // A list of SourceStats
}

/*
 * Service for producing messages
 */
service Producer {
    // Connect to a message broker. Pass in a ConnectionConfiguration with broker
    // specific connection information.
    rpc Connect(arke.ConnectionConfiguration) returns (arke.ConnectResponse) {}
    // Send a stream of messages to the message broker.
    rpc Publish(stream arke.Message) returns (stream arke.MessageResponse) {}
    // Send one message to the message broker.
    rpc PublishOne(arke.Message) returns (arke.MessageResponse) {}
    // Disconnect from the proxy and the message broker.
    rpc Disconnect(arke.Empty) returns (arke.Empty){}
}

/*
 * Service for consuming messages
 */
service Consumer {
    // Connect to a message broker. Pass in a ConnectionConfiguration with broker
    // specific connection information.
    rpc Connect(arke.ConnectionConfiguration) returns (arke.ConnectResponse) {}
    // Subscribe to a message broker source and receive a stream of messages when they are available.
    rpc Consume(stream arke.Consume) returns (stream arke.ConsumeResponse) {}
    // Disconnect from the proxy and the message broker.
    rpc Disconnect(arke.Empty) returns (arke.Empty) {}
    // Ask for and receive information about an arke.Source
    rpc SourceStats(arke.Source) returns (arke.SourceStats) {}
    // Ask for and receive information about a group of arke.Sources
    rpc SourceStatsGroup(arke.Sources) returns (arke.SourceStatsCollection) {}
}

/**
 * Message requesting the health of the other end of the stream. Essentially a ping.
 */
 message HealthCheck {
    string uuid = 1; // UUID to identify the check and match with the response
 }

 /**
 * Message containing the response to a health check request
 */
 message HealthStatus {
    string uuid = 1; // UUID of the HealthCheck message being responded to
    google.protobuf.Timestamp time = 2;
    enum Code {
        OK = 0; // Everything is fine.
        UNHEALTHY = 1; // Everything is not fine (cpu/memory high) but we are operational.
        GOAWAY = 2; // Please go away and come back.
    }
    Code code = 3;
    float cpu_availability = 4; // CPU availability as a ratio (0.0 to 1.0, -1 as unknown)
    float memory_availability = 5; // Memory availability as a ratio (0.0 to 1.0, -1 as unknown)
}

/**
 * Message used to communicate or request health to/from the server and client
 */
message Health {
    oneof resp {
        HealthCheck check = 1;
        HealthStatus status = 2;
    }
}

/**
 * Service for health checks and communication
 */
service Healthz {
    rpc Check(stream arke.Health) returns (stream arke.Health) {}
}