Asynchronous Processing

/**
 * Server-side asynchronous processing context extending AsyncResponse.
 * Provides access to request and response contexts for asynchronous operations.
 */
public interface AsyncContext extends AsyncResponse {
    
    /**
     * Get the container request associated with this async context.
     * @return ContainerRequest for the asynchronous operation
     */
    ContainerRequest getContainerRequest();
    
    /**
     * Get the container response associated with this async context.
     * @return ContainerResponse for the asynchronous operation
     */
    ContainerResponse getContainerResponse();
    
    /**
     * Suspend the request processing.
     * @return true if successfully suspended, false if already suspended
     */
    boolean suspend();
    
    /**
     * Suspend request processing with timeout.
     * @param time Timeout value
     * @param unit Time unit for timeout
     * @return true if successfully suspended
     */
    boolean suspend(long time, TimeUnit unit);
    
    /**
     * Resume processing with a response.
     * @param response Response to resume with
     * @return true if successfully resumed
     */
    boolean resume(Object response);
    
    /**
     * Resume processing with an exception.
     * @param throwable Exception to resume with
     * @return true if successfully resumed
     */
    boolean resume(Throwable throwable);
    
    /**
     * Cancel the asynchronous processing.
     * @return true if successfully cancelled
     */
    boolean cancel();
    
    /**
     * Check if the async processing is suspended.
     * @return true if currently suspended
     */
    boolean isSuspended();
    
    /**
     * Check if the async processing is cancelled.
     * @return true if cancelled
     */
    boolean isCancelled();
}

import org.glassfish.jersey.server.AsyncContext;
import org.glassfish.jersey.server.ContainerRequest;
import jakarta.ws.rs.container.AsyncResponse;
import jakarta.ws.rs.container.Suspended;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

@Path("/async")
public class AsyncResource {
    
    @GET
    @Path("/data")
    public void getDataAsync(@Suspended AsyncResponse asyncResponse) {
        // Cast to AsyncContext for additional functionality
        AsyncContext asyncContext = (AsyncContext) asyncResponse;
        
        // Access request information
        ContainerRequest request = asyncContext.getContainerRequest();
        String userAgent = request.getHeaderString("User-Agent");
        
        // Suspend with timeout
        asyncContext.suspend(30, TimeUnit.SECONDS);
        
        // Perform async operation
        CompletableFuture.supplyAsync(() -> {
            return performLongRunningOperation();
        }).thenAccept(result -> {
            // Resume with result
            asyncContext.resume(result);
        }).exceptionally(throwable -> {
            // Resume with exception
            asyncContext.resume(throwable);
            return null;
        });
    }
    
    @POST
    @Path("/cancel")
    public String cancelAsync(@QueryParam("id") String operationId) {
        AsyncContext context = findAsyncContext(operationId);
        if (context != null && context.isSuspended()) {
            boolean cancelled = context.cancel();
            return cancelled ? "Cancelled" : "Could not cancel";
        }
        return "Operation not found or not suspended";
    }
}

/**
 * Chunked output for streaming responses to clients.
 * Extends GenericType and implements Closeable for resource management.
 */
public class ChunkedOutput<T> extends GenericType<T> implements Closeable {
    
    /**
     * Create chunked output for specified chunk type.
     * @param chunkType Type of chunks to be written
     */
    public ChunkedOutput(Type chunkType);
    
    /**
     * Create chunked output for specified chunk type with separator.
     * @param chunkType Type of chunks to be written
     * @param separator Separator between chunks
     */
    public ChunkedOutput(Type chunkType, String separator);
    
    /**
     * Check if the chunked output is closed.
     * @return true if closed, false otherwise
     */
    public boolean isClosed();
    
    /**
     * Write a chunk to the output.
     * @param chunk Chunk data to write
     * @throws IOException if writing fails or output is closed
     */
    public void write(T chunk) throws IOException;
    
    /**
     * Close the chunked output stream.
     * @throws IOException if closing fails
     */
    public void close() throws IOException;
    
    /**
     * Set exception callback for error handling.
     * @param callback Callback to handle exceptions
     */
    public void setExceptionCallback(ChunkedOutput.ExceptionCallback callback);
    
    /**
     * Exception callback interface for handling write errors.
     */
    public interface ExceptionCallback {
        void onException(ChunkedOutput chunkedOutput, Exception exception);
    }
}

import org.glassfish.jersey.server.ChunkedOutput;
import jakarta.ws.rs.GET;
import jakarta.ws.rs.Path;
import jakarta.ws.rs.Produces;
import jakarta.ws.rs.core.MediaType;
import java.io.IOException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Path("/stream")
public class StreamingResource {
    
    private final ScheduledExecutorService executor = Executors.newScheduledThreadPool(5);
    
    @GET
    @Path("/data")
    @Produces(MediaType.APPLICATION_JSON)
    public ChunkedOutput<String> streamData() {
        ChunkedOutput<String> output = new ChunkedOutput<>(String.class);
        
        // Set exception callback
        output.setExceptionCallback((chunkedOutput, exception) -> {
            System.err.println("Streaming error: " + exception.getMessage());
        });
        
        // Stream data every second for 10 seconds
        executor.scheduleAtFixedRate(() -> {
            try {
                if (!output.isClosed()) {
                    String data = generateData();
                    output.write(data);
                }
            } catch (IOException e) {
                try {
                    output.close();
                } catch (IOException closeEx) {
                    // Log close exception
                }
            }
        }, 0, 1, TimeUnit.SECONDS);
        
        // Auto-close after 10 seconds
        executor.schedule(() -> {
            try {
                output.close();
            } catch (IOException e) {
                // Log exception
            }
        }, 10, TimeUnit.SECONDS);
        
        return output;
    }
    
    @GET
    @Path("/logs")
    @Produces(MediaType.TEXT_PLAIN)
    public ChunkedOutput<String> streamLogs() {
        ChunkedOutput<String> output = new ChunkedOutput<>(String.class, "\n");
        
        // Stream log entries
        streamLogEntries(output);
        
        return output;
    }
    
    private String generateData() {
        return "{\"timestamp\":" + System.currentTimeMillis() + ",\"data\":\"example\"}";
    }
}

/**
 * Broadcaster for sending messages to multiple ChunkedOutput instances.
 * Implements BroadcasterListener for lifecycle management.
 */
public final class Broadcaster<T> implements BroadcasterListener<T> {
    
    /**
     * Create a broadcaster for specific chunk type without automatic closing.
     * @param chunkType Type of chunks to broadcast
     * @return Broadcaster instance
     */
    public static <T> Broadcaster<T> createOnly(Class<T> chunkType);
    
    /**
     * Create a broadcaster for specific chunk type with automatic closing.
     * @param chunkType Type of chunks to broadcast
     * @return Broadcaster instance that auto-closes when empty
     */
    public static <T> Broadcaster<T> create(Class<T> chunkType);
    
    /**
     * Add a chunked output to the broadcaster.
     * @param chunkedOutput ChunkedOutput to add
     * @return true if successfully added
     */
    public boolean add(ChunkedOutput<T> chunkedOutput);
    
    /**
     * Remove a chunked output from the broadcaster.
     * @param chunkedOutput ChunkedOutput to remove
     * @return true if successfully removed
     */
    public boolean remove(ChunkedOutput<T> chunkedOutput);
    
    /**
     * Broadcast a chunk to all registered outputs.
     * @param chunk Chunk to broadcast
     */
    public void broadcast(T chunk);
    
    /**
     * Close all registered outputs and clear the broadcaster.
     */
    public void closeAll();
    
    /**
     * Get the number of registered outputs.
     * @return Number of active chunked outputs
     */
    public int size();
    
    /**
     * Check if the broadcaster is empty.
     * @return true if no outputs are registered
     */
    public boolean isEmpty();
    
    /**
     * Add a broadcaster listener.
     * @param listener Listener to add
     */
    public void addListener(BroadcasterListener<T> listener);
    
    /**
     * Remove a broadcaster listener.
     * @param listener Listener to remove
     */
    public void removeListener(BroadcasterListener<T> listener);
}

import org.glassfish.jersey.server.Broadcaster;
import org.glassfish.jersey.server.ChunkedOutput;
import org.glassfish.jersey.server.BroadcasterListener;
import jakarta.ws.rs.GET;
import jakarta.ws.rs.POST;
import jakarta.ws.rs.Path;
import jakarta.ws.rs.Produces;
import jakarta.ws.rs.core.MediaType;

@Path("/notifications")
public class NotificationResource {
    
    // Static broadcaster for notifications
    private static final Broadcaster<String> broadcaster = Broadcaster.create(String.class);
    
    static {
        // Add listener for broadcaster events
        broadcaster.addListener(new BroadcasterListener<String>() {
            @Override
            public void onClose(ChunkedOutput<String> chunkedOutput) {
                System.out.println("Client disconnected from notifications");
            }
            
            @Override
            public void onException(ChunkedOutput<String> chunkedOutput, Exception exception) {
                System.err.println("Broadcast error: " + exception.getMessage());
            }
        });
    }
    
    @GET
    @Path("/subscribe")
    @Produces(MediaType.TEXT_PLAIN)
    public ChunkedOutput<String> subscribe() {
        ChunkedOutput<String> output = new ChunkedOutput<>(String.class);
        
        // Add to broadcaster
        broadcaster.add(output);
        
        // Send welcome message
        try {
            output.write("Connected to notifications");
        } catch (IOException e) {
            broadcaster.remove(output);
        }
        
        return output;
    }
    
    @POST
    @Path("/send")
    public String sendNotification(String message) {
        // Broadcast to all subscribers
        broadcaster.broadcast("NOTIFICATION: " + message);
        
        return "Notification sent to " + broadcaster.size() + " subscribers";
    }
    
    @GET
    @Path("/stats")
    @Produces(MediaType.APPLICATION_JSON)
    public String getStats() {
        return "{\"subscribers\":" + broadcaster.size() + ",\"isEmpty\":" + broadcaster.isEmpty() + "}";
    }
    
    @POST
    @Path("/shutdown")
    public String shutdown() {
        broadcaster.closeAll();
        return "All notification subscribers disconnected";
    }
}

/**
 * Listener interface for broadcaster events.
 * Provides callbacks for output lifecycle and error handling.
 */
public interface BroadcasterListener<T> {
    
    /**
     * Called when a chunked output is closed.
     * @param chunkedOutput ChunkedOutput that was closed
     */
    default void onClose(ChunkedOutput<T> chunkedOutput) {
        // Default implementation does nothing
    }
    
    /**
     * Called when an exception occurs during broadcasting.
     * @param chunkedOutput ChunkedOutput where exception occurred
     * @param exception Exception that occurred
     */
    default void onException(ChunkedOutput<T> chunkedOutput, Exception exception) {
        // Default implementation does nothing
    }
}

/**
 * Annotation marking methods for managed asynchronous execution.
 * Methods annotated with this will be executed on a managed thread pool.
 */
@Target({ElementType.METHOD, ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
public @interface ManagedAsync {
    // Marker annotation - no parameters
}

/**
 * Qualifier annotation for injecting managed async executor.
 * Used with @Inject to get the managed executor service.
 */
@Qualifier
@Target({ElementType.FIELD, ElementType.PARAMETER, ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
public @interface ManagedAsyncExecutor {
    // Qualifier annotation - no parameters
}

/**
 * Qualifier annotation for injecting background scheduler.
 * Used with @Inject to get the background scheduled executor service.
 */
@Qualifier
@Target({ElementType.FIELD, ElementType.PARAMETER, ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
public @interface BackgroundScheduler {
    // Qualifier annotation - no parameters
}

import org.glassfish.jersey.server.ManagedAsync;
import org.glassfish.jersey.server.ManagedAsyncExecutor;
import org.glassfish.jersey.server.BackgroundScheduler;
import jakarta.inject.Inject;
import jakarta.ws.rs.GET;
import jakarta.ws.rs.Path;
import jakarta.ws.rs.container.AsyncResponse;
import jakarta.ws.rs.container.Suspended;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;

@Path("/managed")
public class ManagedAsyncResource {
    
    @Inject
    @ManagedAsyncExecutor
    private ExecutorService asyncExecutor;
    
    @Inject
    @BackgroundScheduler
    private ScheduledExecutorService scheduler;
    
    @GET
    @Path("/simple")
    @ManagedAsync
    public String getManagedAsync() {
        // This method will be executed on managed thread pool
        return performLongOperation();
    }
    
    @GET
    @Path("/custom")
    public void getCustomAsync(@Suspended AsyncResponse asyncResponse) {
        // Use injected managed executor
        asyncExecutor.submit(() -> {
            try {
                String result = performLongOperation();
                asyncResponse.resume(result);
            } catch (Exception e) {
                asyncResponse.resume(e);
            }
        });
    }
    
    @GET
    @Path("/scheduled")
    public void getScheduledAsync(@Suspended AsyncResponse asyncResponse) {
        // Use injected background scheduler
        scheduler.schedule(() -> {
            try {
                String result = performDelayedOperation();
                asyncResponse.resume(result);
            } catch (Exception e) {
                asyncResponse.resume(e);
            }
        }, 5, TimeUnit.SECONDS);
    }
    
    @ManagedAsync
    @GET
    @Path("/streaming")
    public ChunkedOutput<String> getManagedStreaming() {
        // Managed async with streaming
        ChunkedOutput<String> output = new ChunkedOutput<>(String.class);
        
        scheduler.scheduleAtFixedRate(() -> {
            try {
                if (!output.isClosed()) {
                    output.write("Managed stream data: " + System.currentTimeMillis());
                }
            } catch (IOException e) {
                try {
                    output.close();
                } catch (IOException closeEx) {
                    // Log close exception
                }
            }
        }, 0, 2, TimeUnit.SECONDS);
        
        return output;
    }
    
    private String performLongOperation() {
        // Simulate long-running operation
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return "Long operation completed";
    }
    
    private String performDelayedOperation() {
        return "Delayed operation completed at " + System.currentTimeMillis();
    }
}