proto: support queue

aeon_crud.proto

@@ -43,6 +43,24 @@ service CRUDService {
 // Transactionally executes a set of read and write operations.
 
 message ExecuteRequest {
+  // Message queue push section in a transaction request.
+  message Push {
+    // Definition of messages pushed to a shard-local queue.
+    message Message {
+      // Topic of the message.
+      string topic = 1;
+      // Data of the message.
+      Value data = 2;
+      // Time to live of the message.
+      double ttl = 3;
+    }
+    // Push function.
+    string func = 1;
+    // Additional argument to the push function.
+    Value func_arg = 2;
+    // Messages to send if the push function is not provided.
+    repeated Message messages = 3;
+  }
   // Array of read operations.
   repeated Operation read_set = 1;
   // Array of write operations.
@@ -89,6 +107,8 @@ message ExecuteRequest {
   // Map : space name -> tuple format.
   // Contains formats of all provided tuples. Optional.
   map<string, TupleFormat> tuple_formats = 6;
+  // Description of messages to push when executing a transaction.
+  Push push = 7;
 }
 
 message ExecuteResponse {
@@ -103,6 +123,9 @@ message ExecuteResponse {
   // Map : space name -> tuple format.
   // Contains formats of all returned tuples.
   map<string, TupleFormat> tuple_formats = 5;
+  // PushErr is the error returned by the push function, or nil
+  // if no error occurred.
+  Error push_err = 6;
 }
 
 // Transactionally inserts tuples into a space.

aeon_queue.proto

@@ -0,0 +1,97 @@
+syntax = "proto3";
+
+import "aeon_error.proto";
+import "aeon_value.proto";
+
+package aeon;
+
+// Queue API to Aeon - a distributed database based on Tarantool.
+service QueueService {
+  // Takes messages from a shard-local queue.
+  rpc TakeMessages(TakeMessagesRequest) returns (TakeMessagesResponse) {}
+
+  // Releases messages.
+  rpc ReleaseMessages(ReleaseMessagesRequest)
+      returns (ReleaseMessagesResponse) {}
+
+  // Returns the oldest message for all storages, or the oldest message
+  // for each storage.
+  rpc GetOldestMessages(GetOldestMessagesRequest)
+      returns (GetOldestMessagesResponse) {}
+}
+
+// Description of returned messages.
+message Message {
+  // Shard name.
+  string shard = 1;
+  // The serial number of the message on the shard.
+  uint64 lsn = 2;
+  // Data of the message.
+  Value data = 3;
+}
+
+// Consumer description.
+message ConsumerRef {
+  // Consumer shard.
+  string shard = 1;
+  // Consumer topic.
+  string topic = 2;
+  // Consumer name.
+  string consumer = 3;
+}
+
+message TakeMessagesRequest {
+  // Topic name.
+  string topic = 1;
+  // Unique consumer name.
+  string consumer = 2;
+  // Max number of returned messages.
+  uint64 limit = 3;
+  // Time for the consumer to process the messages.
+  double ttl = 4;
+  // Exclusive mode flag.
+  bool exclusive = 5;
+  // Time to wait for messages.
+  double timeout = 6;
+}
+
+message TakeMessagesResponse {
+  // Error information. Set only on failure.
+  Error error = 1;
+  // Returned messages.
+  repeated Message messages = 2;
+  // Consumer reference used to release messages.
+  ConsumerRef ref = 3;
+  // True if these messages have already been taken by the same consumer,
+  // false otherwise.
+  bool taken_earlier = 4;
+}
+
+message ReleaseMessagesRequest {
+  // Consumer reference.
+  ConsumerRef ref = 1;
+  // If true, released messages will no longer be returned to consumers.
+  bool done = 2;
+}
+
+message ReleaseMessagesResponse {
+  // Error information. Set only on failure.
+  Error error = 1;
+  // True if messages were already released, false otherwise.
+  bool released_earlier = 2;
+}
+
+message GetOldestMessagesRequest {
+  // Topic name.
+  string topic = 1;
+  // True if the oldest messages for each shard should be returned,
+  // false if the oldest messages for all shards should be returned.
+  bool for_each_shard = 2;
+}
+
+message GetOldestMessagesResponse {
+  // Error information. Set only on failure.
+  Error error = 1;
+  // Returned messages.
+  repeated Message messages = 2;
+}