@Override
        public void onSubscribe(Subscription subscription) {
            requireNonNull(subscription, "subscription");
            if (subscribed) {
                subscription.cancel();
                return;
            }

            assert downstream != null;

            subscribed = true;
            final StreamWriter<StreamMessage<HttpData>> newEmitter = newEmitter(subscription);

            // The 'emitter' should be set before reading 'closed' flag.
            // It guarantees that the emitter.abort() or downstream.onError() is always called with
            // 'closedCause'
            emitter = newEmitter;

            if (closed) {
                downstream.onError(closeCause);
                newEmitter.abort(CancelledSubscriptionException.get());
                return;
            }

            final CompletableFuture<Void> completionFuture = MultipartEncoder.this.completionFuture;
            if (completionFuture != null) {
                completeAsync(newEmitter.whenComplete(), completionFuture);
            } else {
                if (!completionFutureUpdater
                        .compareAndSet(MultipartEncoder.this, null, newEmitter.whenComplete())) {
                    final CompletableFuture<Void> oldFuture = MultipartEncoder.this.completionFuture;
                    assert oldFuture != null;
                    completeAsync(newEmitter.whenComplete(), oldFuture);
                }
            }

            StreamMessage.concat(StreamMessage.concat(newEmitter),
                                 StreamMessage.of(HttpData.ofUtf8("--" + boundary + "--")))
                         .subscribe(downstream, executor, options);

            downstream = null;
        }
Reactive processor that encodes a stream of {@link BodyPart}s into an HTTP payload.
core/src/main/java/com/linecorp/armeria/common/multipart/MultipartEncoder.java