@Override
    public void start(Listener<O> responseListener, Metadata metadata) {
        requireNonNull(responseListener, "responseListener");
        requireNonNull(metadata, "metadata");

        // The listener is set by the callers thread. The safe-publishing (aka visibility)
        // to event-loop is guaranteed via semantics of EventLoop.execute(...) call.
        listener = responseListener;

        final Compressor compressor;
        if (callOptions.getCompressor() != null) {
            compressor = compressorRegistry.lookupCompressor(callOptions.getCompressor());
            if (compressor == null) {
                final Status status = Status.INTERNAL
                        .withDescription("Unable to find compressor by name " + callOptions.getCompressor());
                close(status, new Metadata());
                return;
            }
        } else {
            compressor = this.compressor;
        }

        requestFramer.setCompressor(ForwardingCompressor.forGrpc(compressor));

        final long remainingNanos;
        if (callOptions.getDeadline() != null) {
            remainingNanos = callOptions.getDeadline().timeRemaining(TimeUnit.NANOSECONDS);
            if (remainingNanos <= 0) {
                final Status status = Status.DEADLINE_EXCEEDED
                        .augmentDescription("ClientCall started after deadline exceeded: " +
                                            callOptions.getDeadline());
                close(status, new Metadata());
                return;
            } else {
                ctx.setResponseTimeout(TimeoutMode.SET_FROM_NOW, Duration.ofNanos(remainingNanos));
            }
        } else {
            remainingNanos = ctx.remainingTimeoutNanos();
        }

        // Must come after handling deadline.
        final HttpRequest newReq = prepareHeaders(compressor, metadata, remainingNanos);
        final HttpResponse res = ClientUtil.executeWithFallback(preClient, ctx, newReq, errorResponseFactory);

        final HttpStreamDeframer deframer = new HttpStreamDeframer(
                decompressorRegistry, ctx, this, exceptionHandler,
                maxInboundMessageSizeBytes, grpcWebText, false);
        final StreamMessage<DeframedMessage> deframed = res.decode(deframer, ctx.alloc());
        deframer.setDeframedStreamMessage(deframed);

        if (endpointInitialized) {
            deframed.subscribe(this, ctx.eventLoop(), SubscriptionOption.WITH_POOLED_OBJECTS);
        } else {
            addPendingTask(() -> {
                deframed.subscribe(this, ctx.eventLoop(), SubscriptionOption.WITH_POOLED_OBJECTS);
            });
        }

        executor.execute(() -> {
            try (SafeCloseable ignored = ctx.push()) {
                assert listener != null;
                listener.onReady();
            } catch (Throwable t) {
                closeWhenListenerThrows(t);
            }
        });
    }
Encapsulates the state of a single client call, writing messages from the client and reading responses
from the server, passing to business logic via {@link ClientCall.Listener}.
grpc/src/main/java/com/linecorp/armeria/internal/client/grpc/ArmeriaClientCall.java