Broadcasting System

/**
 * Core broadcasting mechanism for message delivery to suspended resources
 */
public interface Broadcaster {
    /**
     * Broadcast message to all connected resources asynchronously
     * @param msg message to broadcast
     * @return Future for async operation
     */
    public Future<Object> broadcast(Object msg);
    
    /**
     * Broadcast message to a specific resource
     * @param msg message to broadcast  
     * @param resource target AtmosphereResource
     * @return Future for async operation
     */
    public Future<Object> broadcast(Object msg, AtmosphereResource resource);
    
    /**
     * Broadcast message to a set of resources
     * @param msg message to broadcast
     * @param resources Set of target AtmosphereResources
     * @return Future for async operation
     */
    public Future<Object> broadcast(Object msg, Set<AtmosphereResource> resources);
    
    /**
     * Schedule delayed broadcast
     * @param msg message to broadcast
     * @param time delay time
     * @param unit TimeUnit for delay
     * @return Future for scheduled operation
     */
    public Future<Object> delayBroadcast(Object msg, long time, TimeUnit unit);
    
    /**
     * Schedule repeated broadcasts at fixed rate
     * @param msg message to broadcast
     * @param period time period between broadcasts
     * @param unit TimeUnit for period
     * @return Future for scheduled operation
     */
    public Future<Object> scheduleFixedBroadcast(Object msg, long period, TimeUnit unit);
    
    /**
     * Add AtmosphereResource to this broadcaster
     * @param resource AtmosphereResource to add
     * @return this broadcaster for chaining
     */
    public Broadcaster addAtmosphereResource(AtmosphereResource resource);
    
    /**
     * Remove AtmosphereResource from this broadcaster
     * @param resource AtmosphereResource to remove
     * @return this broadcaster for chaining
     */
    public Broadcaster removeAtmosphereResource(AtmosphereResource resource);
    
    /**
     * Get all resources associated with this broadcaster
     * @return Collection of AtmosphereResources
     */
    public Collection<AtmosphereResource> getAtmosphereResources();
    
    /**
     * Set the scope for this broadcaster
     * @param scope SCOPE enum (REQUEST, APPLICATION, VM)
     * @return this broadcaster for chaining
     */
    public Broadcaster setScope(SCOPE scope);
    
    /**
     * Get the current scope
     * @return SCOPE enum value
     */
    public SCOPE getScope();
    
    /**
     * Set resource policy for handling resource limits
     * @param policy POLICY enum (FIFO, REJECT)
     * @return this broadcaster for chaining
     */
    public Broadcaster setPolicy(POLICY policy);
    
    /**
     * Get unique identifier for this broadcaster
     * @return broadcaster ID
     */
    public String getID();
    
    /**
     * Destroy this broadcaster and cleanup resources
     */
    public void destroy();
    
    /**
     * Check if broadcaster has been destroyed
     * @return true if destroyed
     */
    public boolean isDestroyed();
    
    /**
     * Add broadcast filter for message processing
     * @param filter BroadcastFilter instance
     * @return this broadcaster for chaining
     */
    public Broadcaster addFilter(BroadcastFilter filter);
    
    /**
     * Remove broadcast filter
     * @param filter BroadcastFilter to remove
     * @return this broadcaster for chaining
     */
    public Broadcaster removeFilter(BroadcastFilter filter);
}

// Get broadcaster and broadcast to all
Broadcaster broadcaster = BroadcasterFactory.getDefault().lookup("/chat", true);
broadcaster.broadcast("Hello everyone!");

// Broadcast to specific resource
broadcaster.broadcast("Private message", specificResource);

// Scheduled broadcasting
broadcaster.scheduleFixedBroadcast("Heartbeat", 30, TimeUnit.SECONDS);

// Add/remove resources
broadcaster.addAtmosphereResource(newResource);
broadcaster.removeAtmosphereResource(disconnectedResource);

// Set scope and policy
broadcaster.setScope(Broadcaster.SCOPE.APPLICATION)
          .setPolicy(Broadcaster.POLICY.FIFO);

/**
 * Default Broadcaster implementation
 */
public class DefaultBroadcaster implements Broadcaster {
    /**
     * Create DefaultBroadcaster with ID
     * @param id unique identifier
     * @param config AtmosphereConfig instance
     */
    public DefaultBroadcaster(String id, AtmosphereConfig config);
    
    /**
     * Set broadcast timeout
     * @param time timeout value
     * @param unit TimeUnit for timeout
     * @return this broadcaster
     */
    public DefaultBroadcaster setTimeout(long time, TimeUnit unit);
    
    /**
     * Set maximum number of suspended resources
     * @param maxSuspendResource maximum resources
     * @return this broadcaster
     */
    public DefaultBroadcaster setMaxSuspendResource(long maxSuspendResource);
}

/**
 * Factory for creating and managing Broadcaster instances
 */
public interface BroadcasterFactory {
    /**
     * Lookup or create broadcaster by ID
     * @param id broadcaster identifier
     * @param createIfNull create if doesn't exist
     * @return Broadcaster instance
     */
    public Broadcaster lookup(Object id, boolean createIfNull);
    
    /**
     * Lookup broadcaster by ID (read-only)
     * @param id broadcaster identifier  
     * @return Broadcaster instance or null
     */
    public Broadcaster lookup(Object id);
    
    /**
     * Create new broadcaster with generated ID
     * @return new Broadcaster instance
     */
    public Broadcaster create();
    
    /**
     * Create new broadcaster with specific ID
     * @param id broadcaster identifier
     * @return new Broadcaster instance
     */
    public Broadcaster create(Object id);
    
    /**
     * Destroy specific broadcaster
     * @param broadcaster Broadcaster to destroy
     * @return true if destroyed
     */
    public boolean destroy(Broadcaster broadcaster);
    
    /**
     * Destroy broadcaster by ID
     * @param id broadcaster identifier
     * @return true if destroyed
     */
    public boolean destroy(Object id);
    
    /**
     * Destroy all broadcasters and cleanup
     * @return this factory
     */
    public BroadcasterFactory destroy();
    
    /**
     * Get all active broadcasters
     * @return Collection of all Broadcasters
     */
    public Collection<Broadcaster> lookupAll();
    
    /**
     * Remove AtmosphereResource from all broadcasters
     * @param resource AtmosphereResource to remove
     * @return this factory
     */
    public BroadcasterFactory removeAllAtmosphereResource(AtmosphereResource resource);
    
    /**
     * Get default factory instance
     * @return default BroadcasterFactory
     */
    public static BroadcasterFactory getDefault();
}

// Get default factory
BroadcasterFactory factory = BroadcasterFactory.getDefault();

// Lookup/create broadcasters
Broadcaster chatBroadcaster = factory.lookup("/chat", true);
Broadcaster newsBroadcaster = factory.create("/news");

// Manage broadcasters
Collection<Broadcaster> allBroadcasters = factory.lookupAll();
factory.destroy(chatBroadcaster);

// Cleanup resources
factory.removeAllAtmosphereResource(disconnectedResource);
factory.destroy(); // Destroy all

/**
 * Allows broadcasting across multiple Broadcasters
 */
public interface MetaBroadcaster {
    /**
     * Broadcast message to specific broadcaster paths
     * @param to path or paths to broadcast to
     * @param message message to broadcast
     * @return MetaBroadcaster for chaining
     */
    public MetaBroadcaster broadcastTo(String to, Object message);
    
    /**
     * Broadcast to multiple broadcaster paths
     * @param to collection of paths
     * @param message message to broadcast
     * @return MetaBroadcaster for chaining
     */
    public MetaBroadcaster broadcastTo(Collection<String> to, Object message);
    
    /**
     * Add broadcaster to meta-broadcaster group
     * @param broadcaster Broadcaster to add
     * @return MetaBroadcaster for chaining
     */
    public MetaBroadcaster addBroadcaster(Broadcaster broadcaster);
    
    /**
     * Remove broadcaster from meta-broadcaster group
     * @param broadcaster Broadcaster to remove
     * @return MetaBroadcaster for chaining
     */
    public MetaBroadcaster removeBroadcaster(Broadcaster broadcaster);
    
    /**
     * Get default MetaBroadcaster instance
     * @return MetaBroadcaster instance
     */
    public static MetaBroadcaster getDefault();
}

// Get meta-broadcaster
MetaBroadcaster metaBroadcaster = MetaBroadcaster.getDefault();

// Broadcast to multiple channels
metaBroadcaster.broadcastTo("/chat", "Message for chat");
metaBroadcaster.broadcastTo("/news", "News update");

// Broadcast to multiple paths at once
List<String> paths = Arrays.asList("/chat", "/news", "/alerts");
metaBroadcaster.broadcastTo(paths, "System announcement");

/**
 * Filter messages before broadcasting
 */
public interface BroadcastFilter {
    /**
     * Filter and potentially transform broadcast message
     * @param originalMessage original message to broadcast
     * @return BroadcastAction containing filtered message and action
     */
    public BroadcastAction filter(Object originalMessage);
    
    /**
     * Filter message for specific resource
     * @param resource target AtmosphereResource
     * @param originalMessage original message
     * @param messageToFilter message being filtered
     * @return BroadcastAction with filtering result
     */
    public BroadcastAction filter(AtmosphereResource resource, Object originalMessage, Object messageToFilter);
}

/**
 * Result of broadcast filtering operation
 */
public static class BroadcastAction {
    /**
     * Action to take with filtered message
     */
    public enum ACTION {
        CONTINUE,  // Continue with broadcast
        ABORT     // Abort broadcast
    }
    
    /**
     * Get the filtered message
     * @return filtered message object
     */
    public Object message();
    
    /**
     * Get the action to take
     * @return ACTION enum value
     */
    public ACTION action();
}

/**
 * Listener for broadcaster events
 */
public interface BroadcasterListener {
    /**
     * Called when resource is added to broadcaster
     * @param broadcaster the Broadcaster
     * @param resource added AtmosphereResource
     */
    public void onAddAtmosphereResource(Broadcaster broadcaster, AtmosphereResource resource);
    
    /**
     * Called when resource is removed from broadcaster
     * @param broadcaster the Broadcaster
     * @param resource removed AtmosphereResource
     */
    public void onRemoveAtmosphereResource(Broadcaster broadcaster, AtmosphereResource resource);
    
    /**
     * Called when broadcaster is destroyed
     * @param broadcaster the destroyed Broadcaster
     */
    public void onDestroy(Broadcaster broadcaster);
    
    /**
     * Called when broadcast completes
     * @param broadcaster the Broadcaster
     * @param message the broadcast message
     * @param resources target resources
     */
    public void onComplete(Broadcaster broadcaster, Object message, Set<AtmosphereResource> resources);
}

/**
 * Broadcaster scope determines resource sharing
 */
public enum SCOPE {
    REQUEST,     // Resources scoped to current request only
    APPLICATION, // Resources shared across web application
    VM          // Resources shared across entire JVM
}

/**
 * Policy for handling resource limits
 */
public enum POLICY {
    FIFO,   // First In, First Out - remove oldest when limit reached
    REJECT  // Reject new resources when limit is reached
}

// Set broadcaster scope
broadcaster.setScope(Broadcaster.SCOPE.APPLICATION);

// Set resource policy  
broadcaster.setPolicy(Broadcaster.POLICY.FIFO);

// Add broadcaster listener
broadcaster.addBroadcasterListener(new BroadcasterListener() {
    @Override
    public void onAddAtmosphereResource(Broadcaster b, AtmosphereResource r) {
        System.out.println("Resource added: " + r.uuid());
    }
    
    @Override
    public void onComplete(Broadcaster b, Object message, Set<AtmosphereResource> resources) {
        System.out.println("Broadcast completed to " + resources.size() + " resources");
    }
});

// Add broadcast filter
broadcaster.addFilter(new BroadcastFilter() {
    @Override
    public BroadcastAction filter(Object originalMessage) {
        // Transform message or abort broadcast
        if (originalMessage == null) {
            return new BroadcastAction(null, BroadcastAction.ACTION.ABORT);
        }
        String filtered = originalMessage.toString().toUpperCase();
        return new BroadcastAction(filtered, BroadcastAction.ACTION.CONTINUE);
    }
});