KurrentDB-Client-Java/src/main/java/io/kurrent/dbclient/AbstractRegularSubscription.java at 73163d42f47b24f10abf0cbb9c444d62fc47058f · kurrent-io/KurrentDB-Client-Java · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
package io.kurrent.dbclient;

import io.kurrent.dbclient.proto.shared.Shared;
import io.kurrent.dbclient.proto.streams.StreamsGrpc;
import io.kurrent.dbclient.proto.streams.StreamsOuterClass;
import io.grpc.ManagedChannel;

import java.util.concurrent.CompletableFuture;

abstract class AbstractRegularSubscription {
    protected static final StreamsOuterClass.ReadReq.Options.Builder defaultReadOptions;
    protected static final StreamsOuterClass.ReadReq.Options.Builder defaultSubscribeOptions;

    protected SubscriptionListener listener;
    protected Checkpointer checkpointer = null;
    private final GrpcClient client;
    private final OptionsWithBackPressureAndSerialization<?> options;

    protected AbstractRegularSubscription(GrpcClient client, OptionsWithBackPressureAndSerialization<?> options) {
        this.client = client;
        this.options = options;
    }

    static {
        defaultReadOptions = StreamsOuterClass.ReadReq.Options.newBuilder()
                .setUuidOption(StreamsOuterClass.ReadReq.Options.UUIDOption.newBuilder()
                        .setStructured(Shared.Empty.getDefaultInstance()));
        defaultSubscribeOptions = defaultReadOptions.clone()
                .setReadDirection(StreamsOuterClass.ReadReq.Options.ReadDirection.Forwards)
                .setSubscription(StreamsOuterClass.ReadReq.Options.SubscriptionOptions.getDefaultInstance());
    }

    protected abstract StreamsOuterClass.ReadReq.Options.Builder createOptions();

    public CompletableFuture<Subscription> execute() {
        return this.client.run(channel -> {
            CompletableFuture<Subscription> future = new CompletableFuture<>();

            StreamsOuterClass.ReadReq readReq = StreamsOuterClass.ReadReq.newBuilder()
                    .setOptions(createOptions())
                    .build();

            StreamsGrpc.StreamsStub streamsClient = GrpcUtils.configureStub(
                    StreamsGrpc.newStub(channel),
                    this.client.getSettings(),
                    this.options
            );

            ReadResponseObserver observer = createObserver(channel, future);
            streamsClient.read(readReq, observer);

            return future;
        });
    }

    private ReadResponseObserver createObserver(ManagedChannel channel, CompletableFuture<Subscription> future) {
        StreamConsumer consumer = new SubscriptionStreamConsumer(
                this.listener,
                this.checkpointer,
                future,
                (subscriptionId, event, action) -> {
                    ClientTelemetry.traceSubscribe(
                            action,
                            subscriptionId,
                            channel,
                            client.getSettings(),
                            options.getCredentials(),
                            event
                    );
                }
        );

        return new ReadResponseObserver(
                this.options,
                consumer,
                this.client.getSerializer(options.serializationSettings().orElse(null))
        );
    }
}