AppendToStream.java

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 com.google.protobuf.ByteString;
import io.grpc.ManagedChannel;
import io.grpc.Metadata;
import io.grpc.StatusRuntimeException;
import io.grpc.stub.StreamObserver;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.CompletableFuture;

class AppendToStream {
    private final GrpcClient client;
    private final String streamName;
    private final StreamState streamState;
    private final List<MessageData> events;
    private final AppendToStreamOptions options;

    public AppendToStream(
            GrpcClient client, 
            String streamName, 
            StreamState streamState, 
            Iterator<MessageData> events, 
            AppendToStreamOptions options
    ) {
        this.client = client;
        this.streamName = streamName;
        this.streamState = streamState;
        this.events = new ArrayList<>();
        while (events.hasNext()) {
            this.events.add(events.next());
        }
        this.options = options;
    }

    public CompletableFuture<WriteResult> execute() {
        return this.client.run(channel -> ClientTelemetry.traceAppend(
                this::append,
                channel,
                events,
                this.streamName,
                this.client.getSettings(),
                this.options.getCredentials()));
    }

    private CompletableFuture<WriteResult> append(ManagedChannel channel, List<MessageData> events) {
        CompletableFuture<WriteResult> result = new CompletableFuture<>();
        StreamsOuterClass.AppendReq.Options.Builder options = this.streamState.applyOnWire(StreamsOuterClass.AppendReq.Options.newBuilder()
                .setStreamIdentifier(Shared.StreamIdentifier.newBuilder()
                        .setStreamName(ByteString.copyFromUtf8(streamName))
                        .build()));
        StreamsGrpc.StreamsStub client = GrpcUtils.configureStub(StreamsGrpc.newStub(channel), this.client.getSettings(), this.options);

        StreamObserver<StreamsOuterClass.AppendReq> requestStream = client.append(GrpcUtils.convertSingleResponse(result, resp -> {
            if (resp.hasSuccess()) {
                StreamsOuterClass.AppendResp.Success success = resp.getSuccess();

                Position logPosition = null;
                if (success.getPositionOptionCase() == StreamsOuterClass.AppendResp.Success.PositionOptionCase.POSITION) {
                    StreamsOuterClass.AppendResp.Position p = success.getPosition();
                    logPosition = new Position(p.getCommitPosition(), p.getPreparePosition());
                }

                StreamState nextExpectedRevision = success.hasNoStream() ? StreamState.noStream()
                        : StreamState.streamRevision(success.getCurrentRevision());

                return new WriteResult(nextExpectedRevision, logPosition);
            }
            if (resp.hasWrongExpectedVersion()) {
                StreamsOuterClass.AppendResp.WrongExpectedVersion wev = resp.getWrongExpectedVersion();

                StreamState expectedRevision;
                if (wev.getExpectedRevisionOptionCase() == StreamsOuterClass.AppendResp.WrongExpectedVersion.ExpectedRevisionOptionCase.EXPECTED_ANY) {
                    expectedRevision = StreamState.any();
                } else if (wev.getExpectedRevisionOptionCase() == StreamsOuterClass.AppendResp.WrongExpectedVersion.ExpectedRevisionOptionCase.EXPECTED_STREAM_EXISTS) {
                    expectedRevision = StreamState.streamExists();
                } else {
                    expectedRevision = StreamState.streamRevision(wev.getExpectedRevision());
                }

                StreamState streamState;
                if (wev.getCurrentRevisionOptionCase() == StreamsOuterClass.AppendResp.WrongExpectedVersion.CurrentRevisionOptionCase.CURRENT_NO_STREAM) {
                    streamState = StreamState.noStream();
                } else {
                    streamState = StreamState.streamRevision(wev.getCurrentRevision());
                }

                String streamName = options.getStreamIdentifier().getStreamName().toStringUtf8();

                throw new WrongExpectedVersionException(streamName, expectedRevision, streamState);
            }

            throw new IllegalStateException("AppendResponse has neither Success or WrongExpectedVersion variants");
        }));

        try {
            requestStream.onNext(StreamsOuterClass.AppendReq.newBuilder().setOptions(options).build());

            for (MessageData e : events) {
                StreamsOuterClass.AppendReq.ProposedMessage.Builder msgBuilder = StreamsOuterClass.AppendReq.ProposedMessage.newBuilder()
                        .setId(Shared.UUID.newBuilder()
                                .setStructured(Shared.UUID.Structured.newBuilder()
                                        .setMostSignificantBits(e.getMessageId().getMostSignificantBits())
                                        .setLeastSignificantBits(e.getMessageId().getLeastSignificantBits())))
                        .setData(ByteString.copyFrom(e.getMessageData()))
                        .putMetadata(SystemMetadataKeys.CONTENT_TYPE, e.getContentType())
                        .putMetadata(SystemMetadataKeys.TYPE, e.getMessageType());

                if (e.getMessageMetadata() != null) {
                    msgBuilder.setCustomMetadata(ByteString.copyFrom(e.getMessageMetadata()));
                }

                requestStream.onNext(StreamsOuterClass.AppendReq.newBuilder()
                        .setProposedMessage(msgBuilder)
                        .build());
            }
            requestStream.onCompleted();
        } catch (StatusRuntimeException e) {
            String leaderHost = e.getTrailers().get(Metadata.Key.of("leader-endpoint-host", Metadata.ASCII_STRING_MARSHALLER));
            String leaderPort = e.getTrailers().get(Metadata.Key.of("leader-endpoint-port", Metadata.ASCII_STRING_MARSHALLER));

            if (leaderHost != null && leaderPort != null) {
                NotLeaderException reason = new NotLeaderException(leaderHost, Integer.parseInt(leaderPort));
                result.completeExceptionally(reason);
            } else {
                result.completeExceptionally(e);
            }
        } catch (RuntimeException e) {
            result.completeExceptionally(e);
        }

        return result;
    }
}