Skip to content

Commit

Permalink
Update to 1.0.0-RC.2 proto (#35)
Browse files Browse the repository at this point in the history
* Update to latest proto from dapr/dapr

* Update signature of publish_event<S> to match PublishEventRequest

* Fix lint message

* Run cargo fmt

* Update example to also print content-type
  • Loading branch information
tcnghia authored Dec 15, 2020
1 parent e764932 commit 49fc694
Show file tree
Hide file tree
Showing 5 changed files with 124 additions and 8 deletions.
8 changes: 4 additions & 4 deletions dapr/proto/runtime/v1/appcallback.proto
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@ service AppCallback {
rpc OnBindingEvent(BindingEventRequest) returns (BindingEventResponse) {}
}

// TopicEventRequest message is compatiable with CloudEvent spec v1.0
// TopicEventRequest message is compatible with CloudEvent spec v1.0
// https://github.com/cloudevents/spec/blob/v1.0/spec.md
message TopicEventRequest {
// id identifies the event. Producers MUST ensure that source + id
Expand Down Expand Up @@ -90,7 +90,7 @@ message TopicEventResponse {

// BindingEventRequest represents input bindings event.
message BindingEventRequest {
// Requried. The name of the input binding component.
// Required. The name of the input binding component.
string name = 1;

// Required. The payload that the input bindings sent
Expand Down Expand Up @@ -142,12 +142,12 @@ message TopicSubscription {
// Required. The name of topic which will be subscribed
string topic = 2;

// The optional properties used for this topic's subscribtion e.g. session id
// The optional properties used for this topic's subscription e.g. session id
map<string,string> metadata = 3;
}

// ListInputBindingsResponse is the message including the list of input bindings.
message ListInputBindingsResponse {
// The list of input bindings.
repeated string bindings = 1;
}
}
90 changes: 90 additions & 0 deletions dapr/proto/runtime/v1/dapr.proto
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ syntax = "proto3";

package dapr.proto.runtime.v1;

import "google/protobuf/any.proto";
import "google/protobuf/empty.proto";
import "dapr/proto/common/v1/common.proto";

Expand Down Expand Up @@ -44,12 +45,27 @@ service Dapr {
// Gets secrets from secret stores.
rpc GetSecret(GetSecretRequest) returns (GetSecretResponse) {}

// Gets a bulk of secrets
rpc GetBulkSecret(GetBulkSecretRequest) returns (GetBulkSecretResponse) {}

// Register an actor timer.
rpc RegisterActorTimer(RegisterActorTimerRequest) returns (google.protobuf.Empty) {}

// Unregister an actor timer.
rpc UnregisterActorTimer(UnregisterActorTimerRequest) returns (google.protobuf.Empty) {}

// Register an actor reminder.
rpc RegisterActorReminder(RegisterActorReminderRequest) returns (google.protobuf.Empty) {}

// Unregister an actor reminder.
rpc UnregisterActorReminder(UnregisterActorReminderRequest) returns (google.protobuf.Empty) {}

// Gets the state for a specific actor.
rpc GetActorState(GetActorStateRequest) returns (GetActorStateResponse) {}

// Executes state transactions for a specified actor
rpc ExecuteActorStateTransaction(ExecuteActorStateTransactionRequest) returns (google.protobuf.Empty) {}

// InvokeActor calls a method on an actor.
rpc InvokeActor (InvokeActorRequest) returns (InvokeActorResponse) {}
}
Expand Down Expand Up @@ -114,6 +130,9 @@ message BulkStateItem {

// The error that was returned from the state store in case of a failed get operation.
string error = 4;

// The metadata which will be sent to app.
map<string,string> metadata = 5;
}

// GetStateResponse is the response conveying the state value and etag.
Expand All @@ -124,6 +143,9 @@ message GetStateResponse {
// The entity tag which represents the specific version of data.
// ETag format is defined by the corresponding data store.
string etag = 2;

// The metadata which will be sent to app.
map<string,string> metadata = 3;
}

// DeleteStateRequest is the message to delete key-value states in the specific state store.
Expand Down Expand Up @@ -165,6 +187,15 @@ message PublishEventRequest {

// The data which will be published to topic.
bytes data = 3;

// The content type for the data (optional).
string data_content_type = 4;

// The metadata passing to pub components
//
// metadata property:
// - key : the key of the message.
map<string,string> metadata = 5;
}

// InvokeBindingRequest is the message to send data to output bindings
Expand Down Expand Up @@ -216,6 +247,22 @@ message GetSecretResponse {
map<string, string> data = 1;
}

// GetBulkSecretRequest is the message to get the secrets from secret store.
message GetBulkSecretRequest {
// The name of secret store.
string store_name = 1;

// The metadata which will be sent to secret store components.
map<string,string> metadata = 2;
}

// GetBulkSecretResponse is the response message to convey the requested secret.
message GetBulkSecretResponse {
// data hold the secret values. Some secret store, such as kubernetes secret
// store, can save multiple secrets for single secret key.
map<string, string> data = 1;
}

// TransactionalStateOperation is the message to execute a specified operation with a key-value pair.
message TransactionalStateOperation {
// The type of operation to be executed
Expand Down Expand Up @@ -255,6 +302,49 @@ message UnregisterActorTimerRequest {
string name = 3;
}

// RegisterActorReminderRequest is the message to register a reminder for an actor of a given type and id.
message RegisterActorReminderRequest {
string actor_type = 1;
string actor_id = 2;
string name = 3;
string due_time = 4;
string period = 5;
bytes data = 6;
}

// UnregisterActorReminderRequest is the message to unregister an actor reminder.
message UnregisterActorReminderRequest {
string actor_type = 1;
string actor_id = 2;
string name = 3;
}

// GetActorStateRequest is the message to get key-value states from specific actor.
message GetActorStateRequest {
string actor_type = 1;
string actor_id = 2;
string key = 3;
}

// GetActorStateResponse is the response conveying the actor's state value.
message GetActorStateResponse {
bytes data = 1;
}

// ExecuteActorStateTransactionRequest is the message to execute multiple operations on a specified actor.
message ExecuteActorStateTransactionRequest {
string actor_type = 1;
string actor_id = 2;
repeated TransactionalActorStateOperation operations = 3;
}

// TransactionalAcorStateOperation is the message to execute a specified operation with a key-value pair.
message TransactionalActorStateOperation {
string operationType = 1;
string key = 2;
google.protobuf.Any value = 3;
}

// InvokeActorRequest is the message to call an actor.
message InvokeActorRequest {
string actor_type = 1;
Expand Down
20 changes: 18 additions & 2 deletions examples/pubsub/publisher.rs
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
use std::{thread, time::Duration};
use std::{collections::HashMap, thread, time::Duration};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
Expand All @@ -16,13 +16,29 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
// name of the pubsub component
let pubsub_name = "pubsub".to_string();

// content type of the pubsub data
let data_content_type = "text/plain".to_string();

// topic to publish message to
let topic = "A".to_string();

for count in 0..100 {
// message metadata
let mut metadata = HashMap::<String, String>::new();
metadata.insert("count".to_string(), count.to_string());

// message
let message = format!("{} => hello from rust!", &count).into_bytes();

client.publish_event(&pubsub_name, &topic, message).await?;
client
.publish_event(
&pubsub_name,
&topic,
&data_content_type,
message,
Some(metadata),
)
.await?;

// sleep for 2 secs to simulate delay b/w two events
tokio::time::delay_for(Duration::from_secs(2)).await;
Expand Down
6 changes: 4 additions & 2 deletions examples/pubsub/subscriber.rs
Original file line number Diff line number Diff line change
Expand Up @@ -40,11 +40,13 @@ impl AppCallback for AppCallbackService {
&self,
request: Request<TopicEventRequest>,
) -> Result<Response<TopicEventResponse>, Status> {
let data = &request.into_inner().data;
let r = request.into_inner();
let data = &r.data;
let data_content_type = &r.data_content_type;

let message = String::from_utf8_lossy(&data);

println!("Message: {}", &message);
println!("Content-Type: {}", &data_content_type);

Ok(Response::new(TopicEventResponse::default()))
}
Expand Down
8 changes: 8 additions & 0 deletions src/client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -86,16 +86,24 @@ impl<T: DaprInterface> Client<T> {
&mut self,
pubsub_name: S,
topic: S,
data_content_type: S,
data: Vec<u8>,
metadata: Option<HashMap<String, String>>,
) -> Result<(), Error>
where
S: Into<String>,
{
let mut mdata = HashMap::<String, String>::new();
if let Some(m) = metadata {
mdata = m;
}
self.0
.publish_event(PublishEventRequest {
pubsub_name: pubsub_name.into(),
topic: topic.into(),
data_content_type: data_content_type.into(),
data,
metadata: mdata,
})
.await
}
Expand Down

0 comments on commit 49fc694

Please sign in to comment.