<commit_before>package io.quarkus.rest.server.runtime.handlers;

import javax.ws.rs.core.MediaType;

import org.jboss.logging.Logger;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import io.quarkus.rest.server.runtime.core.QuarkusRestRequestContext;
import io.quarkus.rest.server.runtime.core.SseUtil;
import io.quarkus.rest.server.runtime.core.StreamingUtil;
import io.quarkus.rest.server.runtime.jaxrs.QuarkusRestOutboundSseEvent;
import io.smallrye.mutiny.Multi;

public class MultiResponseHandler implements ServerRestHandler {

    class SseMultiSubscriber extends AbstractMultiSubscriber {

        SseMultiSubscriber(QuarkusRestRequestContext requestContext) {
            super(requestContext);
        }

        @Override
        public void onNext(Object item) {
            QuarkusRestOutboundSseEvent event = new QuarkusRestOutboundSseEvent.BuilderImpl().data(item).build();
            SseUtil.send(requestContext, event).handle((v, t) -> {
                if (t != null) {
                    // need to cancel because the exception didn't come from the Multi
                    subscription.cancel();
                    handleException(requestContext, t);
                } else {
                    // send in the next item
                    subscription.request(1);
                }
                return null;
            });
        }
    }

    class StreamingMultiSubscriber extends AbstractMultiSubscriber {

        StreamingMultiSubscriber(QuarkusRestRequestContext requestContext) {
            super(requestContext);
        }

        @Override
        public void onNext(Object item) {
            StreamingUtil.send(requestContext, item).handle((v, t) -> {
                if (t != null) {
                    // need to cancel because the exception didn't come from the Multi
                    try {
                        subscription.cancel();
                    } catch (Throwable t2) {
                        t2.printStackTrace();
                    }
                    handleException(requestContext, t);
                } else {
                    // send in the next item
                    subscription.request(1);
                }
                return null;
            });
        }
    }

    abstract class AbstractMultiSubscriber implements Subscriber<Object> {
        protected Subscription subscription;
        protected QuarkusRestRequestContext requestContext;

        AbstractMultiSubscriber(QuarkusRestRequestContext requestContext) {
            this.requestContext = requestContext;
            // let's make sure we never restart by accident, also make sure we're not marked as completed
            requestContext.restart(AWOL);
        }

        @Override
        public void onSubscribe(Subscription s) {
            this.subscription = s;
            // initially ask for one item
            s.request(1);
        }

        @Override
        public void onComplete() {
            // no need to cancel on complete
            // FIXME: are we interested in async completion?
            requestContext.getContext().response().end();
            // so, if I don't also close the connection, the client isn't notified that the request is over
            // I guess that's true of chunked responses, but not clear why I need to close the connection
            // because it means it can't get reused, right?
            requestContext.getContext().response().close();
            requestContext.close();
        }

        @Override
        public void onError(Throwable t) {
            // no need to cancel on error
            handleException(requestContext, t);
        }
    }<commit_msg>SSE: do not lose the target on restart<commit_after>package io.quarkus.rest.server.runtime.handlers;

import javax.ws.rs.core.MediaType;

import org.jboss.logging.Logger;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import io.quarkus.rest.server.runtime.core.QuarkusRestRequestContext;
import io.quarkus.rest.server.runtime.core.SseUtil;
import io.quarkus.rest.server.runtime.core.StreamingUtil;
import io.quarkus.rest.server.runtime.jaxrs.QuarkusRestOutboundSseEvent;
import io.smallrye.mutiny.Multi;

public class MultiResponseHandler implements ServerRestHandler {

    class SseMultiSubscriber extends AbstractMultiSubscriber {

        SseMultiSubscriber(QuarkusRestRequestContext requestContext) {
            super(requestContext);
        }

        @Override
        public void onNext(Object item) {
            QuarkusRestOutboundSseEvent event = new QuarkusRestOutboundSseEvent.BuilderImpl().data(item).build();
            SseUtil.send(requestContext, event).handle((v, t) -> {
                if (t != null) {
                    // need to cancel because the exception didn't come from the Multi
                    subscription.cancel();
                    handleException(requestContext, t);
                } else {
                    // send in the next item
                    subscription.request(1);
                }
                return null;
            });
        }
    }

    class StreamingMultiSubscriber extends AbstractMultiSubscriber {

        StreamingMultiSubscriber(QuarkusRestRequestContext requestContext) {
            super(requestContext);
        }

        @Override
        public void onNext(Object item) {
            StreamingUtil.send(requestContext, item).handle((v, t) -> {
                if (t != null) {
                    // need to cancel because the exception didn't come from the Multi
                    try {
                        subscription.cancel();
                    } catch (Throwable t2) {
                        t2.printStackTrace();
                    }
                    handleException(requestContext, t);
                } else {
                    // send in the next item
                    subscription.request(1);
                }
                return null;
            });
        }
    }

    abstract class AbstractMultiSubscriber implements Subscriber<Object> {
        protected Subscription subscription;
        protected QuarkusRestRequestContext requestContext;

        AbstractMultiSubscriber(QuarkusRestRequestContext requestContext) {
            this.requestContext = requestContext;
            // let's make sure we never restart by accident, also make sure we're not marked as completed
            requestContext.restart(AWOL, true);
        }

        @Override
        public void onSubscribe(Subscription s) {
            this.subscription = s;
            // initially ask for one item
            s.request(1);
        }

        @Override
        public void onComplete() {
            // no need to cancel on complete
            // FIXME: are we interested in async completion?
            requestContext.getContext().response().end();
            // so, if I don't also close the connection, the client isn't notified that the request is over
            // I guess that's true of chunked responses, but not clear why I need to close the connection
            // because it means it can't get reused, right?
            requestContext.getContext().response().close();
            requestContext.close();
        }

        @Override
        public void onError(Throwable t) {
            // no need to cancel on error
            handleException(requestContext, t);
        }
    }
{'fraction_non_alphanumeric': 0.07676821903100255, 'fraction_numerical': 0.0013421017313112334, 'mean_word_length': 1.8562667688769643, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6749893', 'n_tokens_mistral': 1836, 'n_tokens_neox': 1752, 'n_words': 480}