/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you 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 io.statefun.sdk.reqreply;
option java_package = "org.apache.flink.statefun.sdk.reqreply.generated";
option java_multiple_files = true;
// -------------------------------------------------------------------------------------------------------------------
// Common message definitions
// -------------------------------------------------------------------------------------------------------------------
// An Address is the unique identity of an individual StatefulFunction, containing
// a function's type and an unique identifier within the type. The function's
// type denotes the "class" of function to invoke, while the unique identifier addresses the
// invocation to a specific function instance.
message Address {
string namespace = 1;
string type = 2;
string id = 3;
}
message TypedValue {
string typename = 1;
// has_value is set to differentiate a zero length value bytes explicitly set,
// or a non existing value.
bool has_value = 2;
bytes value = 3;
}
// -------------------------------------------------------------------------------------------------------------------
// Messages sent to a Remote Function
// -------------------------------------------------------------------------------------------------------------------
// The following section contains all the message types that are sent
// from Flink to a remote function.
message ToFunction {
// PersistedValue represents a PersistedValue's value that is managed by Flink on behalf of a remote function.
message PersistedValue {
// The unique name of the persisted state.
string state_name = 1;
// The serialized state value
TypedValue state_value = 2;
}
// Invocation represents a remote function call, it associated with an (optional) return address,
// and an argument.
message Invocation {
// The address of the function that requested the invocation (possibly absent)
Address caller = 1;
// The invocation argument (aka the message sent to the target function)
TypedValue argument = 2;
}
// InvocationBatchRequest represents a request to invoke a remote function. It is always associated with a target
// address (the function to invoke), and a list of values for registered state.
message InvocationBatchRequest {
// The address of the function to invoke
Address target = 1;
// A list of PersistedValues that were registered as a persisted state.
repeated PersistedValue state = 2;
// A non empty (at least one) list of invocations
repeated Invocation invocations = 3;
}
oneof request {
InvocationBatchRequest invocation = 100;
}
}
// -------------------------------------------------------------------------------------------------------------------
// Messages sent from a Remote Function
// -------------------------------------------------------------------------------------------------------------------
// The following section contains messages sent from a remote function back to Flink.
message FromFunction {
// MutatePersistedValueCommand represents a command sent from a remote function to Flink,
// requesting a change to a persisted value.
message PersistedValueMutation {
enum MutationType {
DELETE = 0;
MODIFY = 1;
}
MutationType mutation_type = 1;
string state_name = 2;
TypedValue state_value = 3;
}
// Invocation represents a remote function call, it associated with a (mandatory) target address,
// and an argument.
message Invocation {
// The target function to invoke
Address target = 1;
// The invocation argument (aka the message sent to the target function)
TypedValue argument = 2;
}
// DelayedInvocation represents a delayed remote function call with a target address, an argument
// and a delay in milliseconds, after which this message to be sent.
message DelayedInvocation {
// the amount of milliseconds to wait before sending this message
int64 delay_in_ms = 1;
// the target address to send this message to
Address target = 2;
// the invocation argument
TypedValue argument = 3;
}
// EgressMessage an argument to forward to an egress.
// An egress is identified by a namespace and type (see EgressIdentifier SDK class).
// The argument is an io.statefun.sdk.reqreply.TypedValue.
message EgressMessage {
// The target egress namespace
string egress_namespace = 1;
// The target egress type
string egress_type = 2;
// egress argument
TypedValue argument = 3;
}
// InvocationResponse represents a result of an io.statefun.sdk.reqreply.ToFunction.InvocationBatchRequest
// it contains a list of state mutation to preform as a result of computing this batch, and a list of outgoing messages.
message InvocationResponse {
repeated PersistedValueMutation state_mutations = 1;
repeated Invocation outgoing_messages = 2;
repeated DelayedInvocation delayed_invocations = 3;
repeated EgressMessage outgoing_egresses = 4;
}
// ExpirationSpec represents TTL (Time-To-Live) configuration for persisted states.
message ExpirationSpec {
enum ExpireMode {
NONE = 0;
AFTER_WRITE = 1;
AFTER_INVOKE = 2;
}
ExpireMode mode = 1;
int64 expire_after_millis = 2;
}
// PersistedValueSpec represents specifications of a function's persisted value state.
message PersistedValueSpec {
string state_name = 1;
ExpirationSpec expiration_spec = 2;
string type_typename = 3;
}
// IncompleteInvocationContext represents a result of an io.statefun.sdk.reqreply.ToFunction.InvocationBatchRequest,
// which should be used as the response if the InvocationBatchRequest provided incomplete information about the
// invocation, e.g. insufficient state values were provided.
message IncompleteInvocationContext {
repeated PersistedValueSpec missing_values = 1;
}
// Response sent from the function, as a result of an io.statefun.sdk.reqreply.ToFunction.InvocationBatchRequest.
// It can be one of the following types:
// - io.statefun.sdk.reqreply.FromFunction.InvocationResponse
// - io.statefun.sdk.reqreply.FromFunction.IncompleteInvocationContext
oneof response {
InvocationResponse invocation_result = 100;
IncompleteInvocationContext incomplete_invocation_context = 101;
}
}