api.proto

syntax = "proto3";
package proto;


// CreateStreamRequest is sent to create a new stream.
message CreateStreamRequest {
    string subject           = 1; // Stream NATS subject
    string name              = 2; // Stream name (unique per subject)
    string group             = 3; // Partitions NATS subject amongst group members
    int32  replicationFactor = 4; // Number of stream replicas
    int32  partitions        = 5; // Number of stream partitions
}

// CreateStreamResponse is sent by server after creating a stream.
message CreateStreamResponse {
    // Intentionally empty.
}

// StartPosition determines the start-position type on a subscription.
enum StartPosition {
    NEW_ONLY    = 0; // Start at new messages after the latest
    OFFSET      = 1; // Start at a specified offset
    EARLIEST    = 2; // Start at the oldest message
    LATEST      = 3; // Start at the newest message
    TIMESTAMP   = 4; // Start at a specified timestamp
}

// SubscribeRequest is sent to subscribe to a stream partition.
message SubscribeRequest {
    string        stream         = 1; // Stream name to subscribe to
    int32         partition      = 2; // Stream partition to subscribe to
    StartPosition startPosition  = 3; // Where to begin consuming from
    int64         startOffset    = 4; // Offset to begin consuming from
    int64         startTimestamp = 5; // Timestamp to begin consuming from
}

// FetchMetadataRequest is sent to retrieve the latest cluster metadata.
message FetchMetadataRequest {
    repeated string streams = 1; // The streams to fetch metadata for (all if empty)
}

// FetchMetadataResponse contains the cluster metadata requested.
message FetchMetadataResponse {
    repeated Broker         brokers  = 1; // Information for all brokers
    repeated StreamMetadata metadata = 2; // Information for all streams
}

// PublishRequest is sent to publish a new message.
message PublishRequest {
    Message message = 1; // Message to publish
}

// PublishResponse is sent by the server after publishing a message.
message PublishResponse {
    Ack ack = 1; // The ack for the published message if AckPolicy was not NONE
}

// Broker contains information for a Liftbridge broker.
message Broker {
    string id   = 1; // Broker id
    string host = 2; // Broker host
    int32  port = 3; // Broker port
}

// StreamMetadata contains information for a stream.
message StreamMetadata {
    enum Error {
        OK             = 0;
        UNKNOWN_STREAM = 1;
    }
    string                        name       = 1; // The name of the stream being described
    string                        subject    = 2; // The stream subject
    Error                         error      = 3; // Indicates if there was something wrong with the requested stream
    map<int32, PartitionMetadata> partitions = 4; // Information for the stream partitions
}

// PartitionMetadata contains information for a stream partition.
message PartitionMetadata {
    int32            id       = 1; // Partition id
    string           leader   = 2; // Broker id of the partition leader
    repeated string  replicas = 3; // Broker ids of the partition replicas
    repeated string  isr      = 4; // Broker ids of the in-sync replica set
}

// AckPolicy controls the behavior of message acknowledgements.
enum AckPolicy {
    LEADER = 0; // The ack will be sent once the leader has written the message to its log
    ALL    = 1; // The ack will be sent after the ISR replicas have written the message to their logs
    NONE   = 2; // No ack will be sent
}

// Message represents a message from a stream.
message Message {
    int64               offset        = 1;  // Monotonic message offset in the stream
    bytes               key           = 2;  // Message key
    bytes               value         = 3;  // Message payload
    int64               timestamp     = 4;  // When the message was received by the broker
    string              subject       = 5;  // NATS subject message was received on
    string              reply         = 6;  // NATS reply subject
    map<string, bytes>  headers       = 7;  // Message headers
    string              ackInbox      = 8;  // NATS subject to publish acks to
    string              correlationId = 9;  // User-supplied value to correlate acks to publishes
    AckPolicy           ackPolicy     = 10; // Controls the behavior of acks
}

// Ack represents an acknowledgement that a message was committed to a stream
// partition.
message Ack {
    string    stream           = 1; // Name of the stream
    string    partitionSubject = 2; // NATS subject partition is attached to
    string    msgSubject       = 3; // NATS subject the message was received on
    int64     offset           = 4; // Stream offset the message was committed to
    string    ackInbox         = 5; // NATS subject to publish acks to
    string    correlationId    = 6; // User-supplied value from the message
    AckPolicy ackPolicy        = 7; // The AckPolicy sent on the message
}

// API is the main Liftbridge server interface clients interact with.
service API {
    // CreateStream creates a new stream attached to a NATS subject. It returns
    // an AlreadyExists status code if a stream with the given subject and name
    // already exists.
    rpc CreateStream(CreateStreamRequest) returns (CreateStreamResponse) {}

    // Subscribe creates an ephemeral subscription for the given stream. It
    // begins to receive messages starting at the given offset and waits for
    // new messages when it reaches the end of the stream. Use the request
    // context to close the subscription.
    rpc Subscribe(SubscribeRequest) returns (stream Message) {}

    // FetchMetadata retrieves the latest cluster metadata, including stream
    // broker information.
    rpc FetchMetadata(FetchMetadataRequest) returns (FetchMetadataResponse) {}

    // Publish a new message to a subject. If the AckPolicy is not NONE and a
    // deadline is provided, this will synchronously block until the ack is
    // received. If the ack is not received in time, a DeadlineExceeded status
    // code is returned.
    rpc Publish(PublishRequest) returns (PublishResponse) {}
}