class EventStoreClient::GRPC::Client

Public Instance Methods

append_to_stream(stream_name, events, options: {}) click to toggle source

Appends given events to the stream @param [String] Stream name to append events to @param [Array](each: EventStoreClient::DeserializedEvent) list of events to publish @return Dry::Monads::Result::Success or Dry::Monads::Result::Failure

# File lib/event_store_client/adapters/grpc/client.rb, line 12
def append_to_stream(stream_name, events, options: {})
  Commands::Streams::Append.new.call(
    stream_name, events, options: {}
  )
end
delete_stream(stream_name, options: {}) click to toggle source

Softly deletes the given stream @param [String] Stream name to delete @param options [Hash] additional options to the request @return Dry::Monads::Result::Success or Dry::Monads::Result::Failure

# File lib/event_store_client/adapters/grpc/client.rb, line 23
def delete_stream(stream_name, options: {})
  Commands::Streams::Delete.new.call(
    stream_name, options: options
  )
end
listen(subscription, options: {}) { |event| ... } click to toggle source

Runs the persistent subscription indeinitely @param [EventStoreClient::Subscription] subscription to observe @param options [Hash] additional options to the request @return - Nothing, it is a blocking operation, yields the given block with event instead

# File lib/event_store_client/adapters/grpc/client.rb, line 85
def listen(subscription, options: {})
  consume_feed(subscription, options: options) do |event|
    begin
      yield event if block_given?
    rescue StandardError => e
      config.error_handler&.call(e)
    end
  end
end
read(stream_name, options: {}) click to toggle source

Reads a page of events from the given stream @param [String] Stream name to read events from @param options [Hash] additional options to the request @return Dry::Monads::Result::Success with returned events or Dry::Monads::Result::Failure

# File lib/event_store_client/adapters/grpc/client.rb, line 43
def read(stream_name, options: {})
  Commands::Streams::Read.new.call(stream_name, options: options)
end
read_all_from_stream(stream_name, options: {}) click to toggle source

Reads all events from the given stream @param [String] Stream name to read events from @param options [Hash] additional options to the request @return Dry::Monads::Result::Success with returned events or Dry::Monads::Result::Failure

# File lib/event_store_client/adapters/grpc/client.rb, line 52
def read_all_from_stream(stream_name, options: {})
  Commands::Streams::ReadAll.new.call(stream_name, options: options)
end
subscribe(options = {}) { |event| ... } click to toggle source

Subscribe to a stream @param options [Hash] additional options to the request @return - Nothing, it is a blocking operation, yields the given block with event instead

# File lib/event_store_client/adapters/grpc/client.rb, line 99
def subscribe(options = {})
  Commands::Streams::Subscribe.new.call(options) do |event|
    yield event if block_given?
  end
rescue StandardError => e
  config.error_handler&.call(e)
end
subscribe_to_stream(subscription, options: {}) click to toggle source

Creates the subscription for the given stream @param [EventStoreClient::Subscription] subscription to observe @param options [Hash] additional options to the request @return Dry::Monads::Result::Success or Dry::Monads::Result::Failure

# File lib/event_store_client/adapters/grpc/client.rb, line 61
def subscribe_to_stream(subscription, options: {})
  join_streams(subscription.name, subscription.observed_streams)
  Commands::PersistentSubscriptions::Create.new.call(
    subscription.stream,
    subscription.name,
    options: options
  )
end
tombstone_stream(stream_name, options: {}) click to toggle source

Completely removes the given stream @param [String] Stream name to delete @param options [Hash] additional options to the request @return Dry::Monads::Result::Success or Dry::Monads::Result::Failure

# File lib/event_store_client/adapters/grpc/client.rb, line 34
def tombstone_stream(stream_name, options: {})
  Commands::Streams::Tombstone.new.call(stream_name, options: options)
end

Private Instance Methods

consume_feed(subscription, options: {}) { |event| ... } click to toggle source

@api private Consumes the new events from the subscription @param [EventStoreClient::Subscription] subscription to observe @param options [Hash] additional options to the request @return Dry::Monads::Result::Success or Dry::Monads::Result::Failure

# File lib/event_store_client/adapters/grpc/client.rb, line 127
def consume_feed(subscription, options: {})
  Commands::PersistentSubscriptions::Read.new.call(
    subscription.stream, subscription.name, options: options
  ) do |event|
    yield event if block_given?
  end
end
join_streams(name, streams) click to toggle source

Joins multiple streams into the new one under the given name @param [String] Name of the stream containing the ones to join @param [Array] (each: String) list of streams to join together @return Dry::Monads::Result::Success or Dry::Monads::Result::Failure

# File lib/event_store_client/adapters/grpc/client.rb, line 114
def join_streams(name, streams)
  res = Commands::Projections::Create.new.call(name, streams)
  return if res.success?

  Commands::Projections::Update.new.call(name, streams)
end