LiiklusService.proto

syntax = "proto3";

package com.github.bsideup.liiklus;

import "google/protobuf/timestamp.proto";
import "google/protobuf/empty.proto";

option java_package = "com.github.bsideup.liiklus.protocol";
option optimize_for = SPEED;
option java_multiple_files = true;

service LiiklusService {
    rpc Publish (PublishRequest) returns (PublishReply) {

    }

    rpc Subscribe (SubscribeRequest) returns (stream SubscribeReply) {
    }

    rpc Receive (ReceiveRequest) returns (stream ReceiveReply) {

    }

    rpc Ack (AckRequest) returns (google.protobuf.Empty) {

    }

    rpc GetOffsets (GetOffsetsRequest) returns (GetOffsetsReply) {

    }

    rpc GetEndOffsets (GetEndOffsetsRequest) returns (GetEndOffsetsReply) {

    }
}

message LiiklusEvent {
    // Required
    string id = 1;

    // Required
    string type = 2;

    // Required
    string source = 3;

    string time = 5;

    string data_content_type = 100;

    bytes data = 101;

    map<string, string> extensions = 200;
}

message PublishRequest {
    string topic = 1;

    bytes key = 2;

    bytes value = 3 [deprecated = true];

    oneof event {
        LiiklusEvent liiklusEvent = 4;
    }
}

message PublishReply {

    uint32 partition = 1;

    uint64 offset = 2;

    string topic = 3;
}

message SubscribeRequest {

    string topic = 1;

    string group = 2;

    uint32 group_version = 4;

    AutoOffsetReset auto_offset_reset = 3;

    enum AutoOffsetReset {
        EARLIEST = 0;
        LATEST = 1;
    }
}

message Assignment {
    string session_id = 1;

    uint32 partition = 2;
}

message SubscribeReply {
    oneof reply {
        Assignment assignment = 1;
    }
}

message AckRequest {
    Assignment assignment = 1 [deprecated = true];

    string topic = 3;
    string group = 4;
    uint32 group_version = 5;
    uint32 partition = 6;

    uint64 offset = 2;
}

message ReceiveRequest {
    Assignment assignment = 1;

    uint64 last_known_offset = 2;

    ContentFormat format = 3;

    enum ContentFormat {
        BINARY = 0;
        LIIKLUS_EVENT = 1;
    }
}

message ReceiveReply {
    oneof reply {
        Record record = 1;
        LiiklusEventRecord liiklus_event_record = 2;
    }

    message Record {
        uint64 offset = 1;

        bytes key = 2;

        bytes value = 3;

        google.protobuf.Timestamp timestamp = 4;

        bool replay = 5;
    }

    message LiiklusEventRecord {
        uint64 offset = 1;

        bytes key = 2;

        LiiklusEvent event = 3;

        // TODO drop?
        google.protobuf.Timestamp timestamp = 4;

        bool replay = 5;
    }
}

message GetOffsetsRequest {

    string topic = 1;

    string group = 2;

    uint32 group_version = 3;

}

message GetOffsetsReply {
    map<uint32, uint64> offsets = 1;
}

message GetEndOffsetsRequest {
    string topic = 1;
}

message GetEndOffsetsReply {
    map<uint32, uint64> offsets = 1;
}