Embedded client AccessDeniedException subscription to all

Clients
1

I am using the EventStore.Client.Embedded Version=20.6.0 and it’s all good for appending events to stream and reading, but when I try to subscribe it complains about access denied Unhandled exception. EventStore.ClientAPI.Exceptions.AccessDeniedException: Subscription to '<all>' failed due to access denied.
and I’m not sure why because I am using it in-memory.

This is my code, it uses the IEventStoreConnection to subscribe to all streams and then it ignores all the events that don’t belong in an expected stream

public class EventStoreSubscriber
    : IEventStoreSubscriber
{
    private readonly IEventStoreConnection _eventStoreConnection;
    private readonly SerializationService _serializationService;

    public EventStoreSubscriber(
        IEventStoreConnection eventStoreConnection,
        SerializationService serializationService)
    {
        _eventStoreConnection = eventStoreConnection;
        _serializationService = serializationService;
    }

    public Task Subscribe(string streamNamePrefix, Func<object, CancellationToken, Task> handler, CancellationToken cancellationToken)
    {
        return _eventStoreConnection.SubscribeToAllAsync(
            false,
            async (streamSubscription, resolvedEvent) =>
            {
                var streamName = resolvedEvent.OriginalStreamId;
                if (streamName.StartsWith(streamNamePrefix, StringComparison.InvariantCultureIgnoreCase))
                {
                    var recognizedEvent = _serializationService.GetAsEvent(resolvedEvent);
                    await handler(recognizedEvent, cancellationToken);
                }
            });
    }
}

This approach works well with the “real” event store. What should I do to have the embedded ES also work with this approach?

This is how I use it

var nodeBuilder =
    EmbeddedVNodeBuilder
        .AsSingleNode()
        .OnDefaultEndpoints()
        .RunInMemory();

var node = nodeBuilder.Build();
await node.StartAsync(true);
using var embeddedConnection = EmbeddedEventStoreConnection.Create(node);
var supportedTypes = new SupportedTypes();
supportedTypes.AddType(typeof(Foo));
supportedTypes.AddType(typeof(Bar));
var serializationService = new SerializationService(supportedTypes);
var eventStoreSubscriber = new EventStoreSubscriber(embeddedConnection, serializationService);

await eventStoreSubscriber.Subscribe(
    streamNamePrefix, async (retrievedEvent, cancellationToken) =>
    {
        await Console.Out.WriteLineAsync($"Subscriber {subscriberId} received: {retrievedEvent}");
    },
    CancellationToken.None
);
2

@Joao_Braganca do have any ideas?

3

Did you change the default ACLs or turn off security? If not you still need to pass in a user that can read from $all.

4

Didn’t know about those things, thanks. I managed to add a user

var nodeBuilder =
            EmbeddedVNodeBuilder
                .AsSingleNode()
                .OnDefaultEndpoints()
                .WithEnableAtomPubOverHTTP(true)
                .RunInMemory();

var node = nodeBuilder.Build();
        await node.StartAsync(true);

var connectionSettings =
            ConnectionSettings
                .Create()
                .SetDefaultUserCredentials(new UserCredentials("admin", "changeit"))
                .KeepReconnecting();
        
using var embeddedConnection = EmbeddedEventStoreConnection.Create(node,connectionSettings);

I don’t get the AccessDeniedException anymore, but the subscribers don’t pick any event like this.

await embeddedConnection.SubscribeToAllAsync(false,
            async (streamSubscription, resolvedEvent) =>
                await Console.Out.WriteLineAsync(resolvedEvent.ToString()));

Any idea what could be the reason or how could I troubleshoot?
If I read events from the stream I can find them there, so there’s definitely a stream with events.

Thanks again!

5

As you are using IEventStoreConnection, SubscribeToAll is live and will only pick up new events. What you want is SubscribeToAllFrom, where you can pass in a last known position (or null) to start from the beginning.