persistent subscriptions to :all

[the-mikedavis]

EventStore/EventStore#2869 was recently merged that added server-side support for persistent subscriptions to the magic $all stream.

The syntax in Spear should be roughly the same as it works for Spear.subscribe/4: pass the :all atom instead of a stream name to use the all stream

iex> Spear.connect_to_persistent_subscription(conn, self(), :all, "PSubIEx")

We'll have to wait until a stable version of EventStoreDB is released that supports this version, though.

[the-mikedavis]

looks like v21.6.0 was just released today with this feature https://www.eventstore.com/blog/21.6.0-release-notes

[the-mikedavis]

here are the relevant changes to the persistent.proto definitions:

diff --git a/src/Protos/Grpc/persistent.proto b/src/Protos/Grpc/persistent.proto
index 04243ad03..44a047235 100644
--- a/src/Protos/Grpc/persistent.proto
+++ b/src/Protos/Grpc/persistent.proto
@@ -19,27 +19,31 @@ message ReadReq {
 	}
 
 	message Options {
-		event_store.client.shared.StreamIdentifier stream_identifier = 1;
+		oneof stream_option {
+			event_store.client.StreamIdentifier stream_identifier = 1;
+			event_store.client.Empty all = 5;
+		}
+
 		string group_name = 2;
 		int32 buffer_size = 3;
 		UUIDOption uuid_option = 4;
@@ -89,14 +93,60 @@ message CreateReq {
 	Options options = 1;
 
 	message Options {
-		event_store.client.shared.StreamIdentifier stream_identifier = 1;
+		oneof stream_option {
+			StreamOptions stream = 4;
+			AllOptions all = 5;
+		}
+		event_store.client.StreamIdentifier stream_identifier = 1 [deprecated=true];
 		string group_name = 2;
 		Settings settings = 3;
 	}
 
+	message StreamOptions {
+		event_store.client.StreamIdentifier stream_identifier = 1;
+		oneof revision_option {
+			uint64 revision = 2;
+			event_store.client.Empty start = 3;
+			event_store.client.Empty end = 4;
+		}
+	}
+
+	message AllOptions {
+		oneof all_option {
+			Position position = 1;
+			event_store.client.Empty start = 2;
+			event_store.client.Empty end = 3;
+		}
+		oneof filter_option {
+			FilterOptions filter = 4;
+			event_store.client.Empty no_filter = 5;
+		}
+		message FilterOptions {
+			oneof filter {
+				Expression stream_identifier = 1;
+				Expression event_type = 2;
+			}
+			oneof window {
+				uint32 max = 3;
+				event_store.client.Empty count = 4;
+			}
+			uint32 checkpointIntervalMultiplier = 5;
+
+			message Expression {
+				string regex = 1;
+				repeated string prefix = 2;
+			}
+		}
+	}
+
+	message Position {
+		uint64 commit_position = 1;
+		uint64 prepare_position = 2;
+	}
+
 	message Settings {
 		bool resolve_links = 1;
-		uint64 revision = 2;
+		uint64 revision = 2 [deprecated = true];
 		bool extra_statistics = 3;
 		int32 max_retry_count = 5;
 		int32 min_checkpoint_count = 7;
@@ -130,14 +180,40 @@ message UpdateReq {
 	Options options = 1;
 
 	message Options {
-		event_store.client.shared.StreamIdentifier stream_identifier = 1;
+		oneof stream_option {
+			StreamOptions stream = 4;
+			AllOptions all = 5;
+		}
+		event_store.client.StreamIdentifier stream_identifier = 1 [deprecated = true];
 		string group_name = 2;
 		Settings settings = 3;
 	}
 
+	message StreamOptions {
+		event_store.client.StreamIdentifier stream_identifier = 1;
+		oneof revision_option {
+			uint64 revision = 2;
+			event_store.client.Empty start = 3;
+			event_store.client.Empty end = 4;
+		}
+	}
+
+	message AllOptions {
+		oneof all_option {
+			Position position = 1;
+			event_store.client.Empty start = 2;
+			event_store.client.Empty end = 3;
+		}
+	}
+
+	message Position {
+		uint64 commit_position = 1;
+		uint64 prepare_position = 2;
+	}
+
 	message Settings {
 		bool resolve_links = 1;
-		uint64 revision = 2;
+		uint64 revision = 2 [deprecated = true];
 		bool extra_statistics = 3;
 		int32 max_retry_count = 5;
 		int32 min_checkpoint_count = 7;
@@ -171,7 +247,11 @@ message DeleteReq {
 	Options options = 1;
 
 	message Options {
-		event_store.client.shared.StreamIdentifier stream_identifier = 1;
+		oneof stream_option {
+			event_store.client.StreamIdentifier stream_identifier = 1;
+			event_store.client.Empty all = 3;
+		}
+
 		string group_name = 2;
 	}
 }

(note that this update to the protos also changed out event_store.client.{shared. => }* and those changes have been omitted from the above)

[yordis]

Hey there, any updates on this one? How can I help?

[the-mikedavis]

Hey Yordis,

this one fell by the way-side a bit but I don't think it's a huge amount of work. In theory this should be a pretty small change, just implementing this block with the new protobuf definitions:

spear/lib/spear/persistent_subscription.ex

Lines 135 to 138 in 8a1d13f

	def map_create_stream_option(:all, _opts) do
	# YARD
	{:all, Persistent.create_req_all_options()}
	end

, but I think there'll be some reorganization of the persistent subscription code along with it to make things a bit more clean. I'll take a look at it this afternoon 👍

[the-mikedavis]

@yordis I'm gonna publish the new changes so far (see the CHANGELOG section on v0.10.0) as v0.10.0-rc.1 so you can try it out without taking on a git dependency. Ideally I'd like to roll #53 into the v0.10.0 release but if you'd like a real release for this feature before then, give me a shout