// Copyright 2020 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

syntax = "proto3";

package google.cloud.pubsublite.v1;

import "google/cloud/pubsublite/v1/common.proto";
import "google/api/client.proto";

option cc_enable_arenas = true;
option go_package = "google.golang.org/genproto/googleapis/cloud/pubsublite/v1;pubsublite";
option java_multiple_files = true;
option java_outer_classname = "SubscriberProto";
option java_package = "com.google.cloud.pubsublite.proto";

// The service that a subscriber client application uses to receive messages
// from subscriptions.
service SubscriberService {
  option (google.api.default_host) = "pubsublite.googleapis.com";
  option (google.api.oauth_scopes) = "https://www.googleapis.com/auth/cloud-platform";

  // Establishes a stream with the server for receiving messages.
  rpc Subscribe(stream SubscribeRequest) returns (stream SubscribeResponse) {
  }
}

// The first request that must be sent on a newly-opened stream. The client must
// wait for the response before sending subsequent requests on the stream.
message InitialSubscribeRequest {
  // The subscription from which to receive messages.
  string subscription = 1;

  // The partition from which to receive messages. Partitions are zero indexed,
  // so `partition` must be in the range [0, topic.num_partitions).
  int64 partition = 2;
}

// Response to an InitialSubscribeRequest.
message InitialSubscribeResponse {
  // The cursor from which the subscriber will start receiving messages once
  // flow control tokens become available.
  Cursor cursor = 1;
}

// Request to update the stream's delivery cursor based on the given target.
// Resets the server available tokens to 0. SeekRequests may not be sent while
// another SeekRequest is outstanding (i.e., has not received a SeekResponse) on
// the same stream. SeekRequests past head result in stream breakage.
message SeekRequest {
  // A special target in the partition that takes no other parameters.
  enum NamedTarget {
    // Default value. This value is unused.
    NAMED_TARGET_UNSPECIFIED = 0;

    // A target corresponding to the most recently published message in the
    // partition.
    HEAD = 1;

    // A target corresponding to the committed cursor for the given subscription
    // and topic partition.
    COMMITTED_CURSOR = 2;
  }

  // The target to seek to. Must be set.
  oneof target {
    // A named target.
    NamedTarget named_target = 1;

    // A target corresponding to the cursor, pointing to anywhere in the
    // topic partition.
    Cursor cursor = 2;
  }
}

// Response to a SeekRequest.
message SeekResponse {
  // The new delivery cursor for the current stream.
  Cursor cursor = 1;
}

// Request to grant tokens to the server, requesting delivery of messages when
// they become available.
message FlowControlRequest {
  // The number of message tokens to grant. Must be greater than or equal to 0.
  int64 allowed_messages = 1;

  // The number of byte tokens to grant. Must be greater than or equal to 0.
  int64 allowed_bytes = 2;
}

// A request sent from the client to the server on a stream.
message SubscribeRequest {
  oneof request {
    // Initial request on the stream.
    InitialSubscribeRequest initial = 1;

    // Request to update the stream's delivery cursor.
    SeekRequest seek = 2;

    // Request to grant tokens to the server,
    FlowControlRequest flow_control = 3;
  }
}

// Response containing a list of messages. Upon delivering a MessageResponse to
// the client, the server:
// *  Updates the stream's delivery cursor to one greater than the cursor of the
//    last message in the list.
// *  Subtracts the total number of bytes and messages from the tokens available
//    to the server.
message MessageResponse {
  // Messages from the topic partition.
  repeated SequencedMessage messages = 1;
}

// Response to SubscribeRequest.
message SubscribeResponse {
  oneof response {
    // Initial response on the stream.
    InitialSubscribeResponse initial = 1;

    // Response to a Seek operation.
    SeekResponse seek = 2;

    // Response containing messages from the topic partition.
    MessageResponse messages = 3;
  }
}