#126 - add initialDelay to SSE scheduler to give server's handshake processing time

[xsalefter]

Original ticket : #126 . This PR also follow-up for this PR in Kill Bill repository : killbill/killbill#2169

LogsSseHandler scheduled its drain loop with initialDelay=0, so the background thread could publish events while the servlet container was still completing the SSE handshake. In that window the channel remains in its blocking phase (asyncStarted could be true, but the response isn’t committed yet), and the container throw exception (IllegalStateException: api=BLOCKED in Jetty) when we write.

Video below shows what exactly happened:

1. 00.00 : Start server and initialize 2 curl calls to SSE endpoint
2. 00.42 : Show log for second SSE request with jettyState=HANDLING .... responseCommitted=false meaning the server still processing the handshake/request, and response not committed yet.

platform-126-SSE-without-initial-delay.webm

To be continued in comment ....

[xsalefter]

Now this video show what happened if we set the initialDelay (as we did in this PR):

1. 00.00 : Server started
2. 00.16 : Initialize 2 curl calls to SSE endpoint. Inside LogsSseHandler's executorService The server (jetty in this example) state now consistent with jettyState=WAITING, asyncStarted=true, responseCommitted=true (different with previous state where first log contains jettyState=HANDLING, asyncStarted=true, responseCommitted=false)
3. 00.35 : Create another SSE request. No error happened.

platform-126-SSE-with-initial-delay.webm

[xsalefter]

Add compile dependency to jetty (for debugging purpose) raise another OSGi issue. So I have to use reflection to get Jetty classes. Also, using debugger and set breakpoint is not working as well (I never get the expected exception). Probably because the server already have in correct state when the line of code hit breakpoint.

    private void debug(final Sse sse) {
        if (!(sse instanceof ServletSse)) {
            logger.info(">>>>> SSE instance is not ServletSse: {}", sse.getClass().getName());
            return;
        }

        try {
            final ServletSse servletSse = (ServletSse) sse;
            final Field reqField = ServletSse.class.getDeclaredField("req");
            reqField.setAccessible(true);
            HttpServletRequest rawRequest = (HttpServletRequest) reqField.get(servletSse);
            while (rawRequest instanceof HttpServletRequestWrapper) {
                rawRequest = (HttpServletRequest) ((HttpServletRequestWrapper) rawRequest).getRequest();
            }

            final Object state = extractChannelState(rawRequest);
            if (state == null) {
                logger.info(">>>>> Jetty HttpChannelState unavailable for {} (loader {})",
                            rawRequest.getClass().getName(),
                            rawRequest.getClass().getClassLoader());
                return;
            }

            logChannelState(state);
        } catch (final ReflectiveOperationException e) {
            logger.warn(">>>>> Failed to inspect Jetty request state", e);
        } catch (final Exception e) {
            logger.warn(">>>>> Unexpected failure while inspecting Jetty request state", e);
        }
    }

    private Object extractChannelState(final HttpServletRequest rawRequest) throws ReflectiveOperationException {
        // First, try to interact with the server's Jetty classes via reflection to avoid class loader mismatches.
        final Class<?> requestClass = rawRequest.getClass();
        try {
            final java.lang.reflect.Method method = requestClass.getMethod("getHttpChannelState");
            return method.invoke(rawRequest);
        } catch (final NoSuchMethodException ignored) {
            // fall back to Jetty API if the method isn't present
        }

        // If the container request exposes Jetty 10 APIs, use them directly.
        try {
            final Class<?> jettyRequestClass = Class.forName("org.eclipse.jetty.server.Request", false, requestClass.getClassLoader());
            if (jettyRequestClass.isInstance(rawRequest)) {
                final java.lang.reflect.Method getChannelState = jettyRequestClass.getMethod("getHttpChannelState");
                return getChannelState.invoke(rawRequest);
            }
        } catch (final ClassNotFoundException cnfe) {
            // Jetty classes are not present in this class loader
            logger.error(">>>>> Jetty Request class not found in class loader: ", cnfe);
        }

        return null;
    }

    private void logChannelState(final Object state) throws ReflectiveOperationException {
        final Class<?> stateClass = state.getClass();

        final java.lang.reflect.Method getState = stateClass.getMethod("getState");
        final java.lang.reflect.Method isAsyncStarted = stateClass.getMethod("isAsyncStarted");
        final java.lang.reflect.Method isResponseCommitted = stateClass.getMethod("isResponseCommitted");

        logger.info(">>>>> jettyState={}, asyncStarted={}, responseCommitted={}",
                    getState.invoke(state),
                    isAsyncStarted.invoke(state),
                    isResponseCommitted.invoke(state));
    }