Subscribe only to a single stream

[aef]

I asked my question over at StackOverflow.

Would be very nice if you had some answers for me.

Thanks.

[aef]

Can it really be that there is no way to do this? Why have separate streams at all if you can't subscribe to them separately?

[jandudulski]

Why have separate streams at all if you can't subscribe to them separately?

To make it efficient to load events for specific aggregate (so it doesn't need to filter millions of events on every call).

Can it really be that there is no way to do this?

The idea is that you subscribe to events, not streams. If you would subscribe to streams than you would need to create subscriptions dynamically and that would increase complexity a lot. It's much simpler to just filter by event attributes.

And I think it applies to probably every implementation of event store as defined by Greg Young.

[mpraglowski]

Greg's EventStore (mentioned here) implements this feature with server-side projections - https://eventstore.org/docs/projections/user-defined-projections/index.html

RES does not have server side projections implemented. You need to implement the projection by yourself. There is no way to subscribe for events from selected stream (however the idea seems interesting). You need to implement your builder to read from the given stream and process events periodically. It need to store last processed event id and next time it runs it should read only new domain events linked to stream.

The (pseudo) code could look like:

class Builder
  def initialize(event_store)
    @event_store = event_store
    @last_processed_event_id = read_last_processed_event
  end

  def call(stream)
    events = read_all_since(stream, event_id)
    process(events)
    store_last_processed_event
  end
  
  private
  def read_all_since(stream, id)
    @event_store.read.stream(stream).from(id).to_a
  end

  def process(events)
    events.each do |event|
      #... do sth here
      @last_processed_event_id = event.event_id
    end
  end

  def load_last_processed_event
    # read last processed event or return nil to start from beginning of stream
  end

  def store_last_processed_event
    # store it somewhere
  end
end