Embedded Broker

/**
 * Main broker service implementation for embedded brokers
 * Provides full broker functionality within applications
 */
public class BrokerService implements Service {
    /** Create new broker service */
    public BrokerService();
    
    /** Start broker service */
    public void start() throws Exception;
    
    /** Stop broker service */
    public void stop() throws Exception;
    
    /** Broker identification */
    public void setBrokerName(String brokerName);
    public String getBrokerName();
    public String getBrokerId();
    
    /** Transport connectors */
    public TransportConnector addConnector(String bindAddress) throws Exception;
    public TransportConnector addConnector(URI bindAddress) throws Exception;
    public void removeConnector(TransportConnector connector) throws Exception;
    public TransportConnector[] getTransportConnectors();
    public void setTransportConnectors(List<TransportConnector> transportConnectors) throws Exception;
    
    /** Network connectors for broker clustering */
    public NetworkConnector addNetworkConnector(String discoveryAddress) throws Exception;
    public NetworkConnector addNetworkConnector(URI discoveryAddress) throws Exception;
    public void removeNetworkConnector(NetworkConnector connector) throws Exception;
    public NetworkConnector[] getNetworkConnectors();
    public void setNetworkConnectors(List<NetworkConnector> networkConnectors) throws Exception;
    
    /** Persistence configuration */
    public void setPersistenceAdapter(PersistenceAdapter persistenceAdapter);
    public PersistenceAdapter getPersistenceAdapter();
    public void setPersistent(boolean persistent);
    public boolean isPersistent();
    
    /** Data directory for persistent storage */
    public void setDataDirectory(String dataDirectory);
    public String getDataDirectory();
    public void setDataDirectoryFile(File dataDirectoryFile);
    public File getDataDirectoryFile();
    
    /** JMX management */
    public void setUseJmx(boolean useJmx);
    public boolean isUseJmx();
    public void setJmxDomainName(String jmxDomainName);
    public String getJmxDomainName();
    public ManagementContext getManagementContext();
    public void setManagementContext(ManagementContext managementContext);
    
    /** Security configuration */
    public void setPlugins(BrokerPlugin[] plugins);
    public BrokerPlugin[] getPlugins();
    public void setSecurityManager(SecurityManager securityManager);
    
    /** Destination policies */
    public void setDestinationPolicy(PolicyMap destinationPolicy);
    public PolicyMap getDestinationPolicy();
    
    /** System usage limits */
    public void setSystemUsage(SystemUsage systemUsage);
    public SystemUsage getSystemUsage();
    
    /** Broker configuration */
    public void setPopulateJMSXUserID(boolean populateJMSXUserID);
    public boolean isPopulateJMSXUserID();
    public void setUseAuthenticatedPrincipalForJMSXUserID(boolean useAuthenticatedPrincipalForJMSXUserID);
    public boolean isUseAuthenticatedPrincipalForJMSXUserID();
    
    /** Advisory message support */
    public void setAdvisorySupport(boolean advisorySupport);
    public boolean isAdvisorySupport();
    
    /** Administrative view */
    public Broker getBroker();
    public BrokerView getAdminView() throws Exception;
    
    /** Wait for broker to stop */
    public void waitUntilStopped();
    public boolean waitUntilStarted();
    public boolean waitUntilStarted(int timeout);
}

// Basic embedded broker
BrokerService broker = new BrokerService();
broker.setBrokerName("embedded-broker");
broker.addConnector("tcp://localhost:61616");
broker.setPersistent(false); // Use memory persistence
broker.start();

// Use the broker...
ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");

// Stop broker
broker.stop();

// Broker with persistence
BrokerService persistentBroker = new BrokerService();
persistentBroker.setBrokerName("persistent-broker");
persistentBroker.setDataDirectory("./activemq-data");
persistentBroker.addConnector("tcp://localhost:61617");
persistentBroker.start();

// Broker with multiple connectors
BrokerService multiBroker = new BrokerService();
multiBroker.setBrokerName("multi-transport-broker");
multiBroker.addConnector("tcp://localhost:61616");
multiBroker.addConnector("stomp://localhost:61613");
multiBroker.addConnector("mqtt://localhost:1883");
multiBroker.start();

/**
 * Factory for creating brokers from URI configurations
 */
public class BrokerFactory {
    /** Create broker from URI configuration */
    public static BrokerService createBroker(URI config) throws Exception;
    
    /** Create broker from URI string */
    public static BrokerService createBroker(String uri) throws Exception;
    
    /** Create broker and start it */
    public static BrokerService createBrokerFromURI(String uri) throws Exception;
    
    /** Register broker factory handler */
    public static void registerBrokerFactoryHandler(String scheme, BrokerFactoryHandler handler);
    
    /** Get broker factory handler */
    public static BrokerFactoryHandler createBrokerFactoryHandler(String type);
}

/**
 * Interface for broker factory handlers
 */
public interface BrokerFactoryHandler {
    /** Create broker from URI */
    BrokerService createBroker(URI uri) throws Exception;
}

// Create broker from URI
BrokerService broker = BrokerFactory.createBroker("broker:(tcp://localhost:61616)?persistent=false");
broker.start();

// Create and start broker in one step
BrokerService autoStartBroker = BrokerFactory.createBrokerFromURI(
    "broker:(tcp://localhost:61616,stomp://localhost:61613)?persistent=false"
);

// VM transport for embedded usage
BrokerService vmBroker = BrokerFactory.createBroker("broker:vm://embedded?persistent=false");

/**
 * Transport connector for accepting client connections
 */
public class TransportConnector implements Service {
    /** Start accepting connections */
    public void start() throws Exception;
    
    /** Stop accepting connections */
    public void stop() throws Exception;
    
    /** Connector configuration */
    public void setUri(URI uri);
    public URI getUri();
    public void setName(String name);
    public String getName();
    
    /** Connection limits */
    public void setMaximumConnections(int maximumConnections);
    public int getMaximumConnections();
    public int getConnectionCount();
    
    /** Security configuration */
    public void setDiscoveryAgent(DiscoveryAgent discoveryAgent);
    public DiscoveryAgent getDiscoveryAgent();
    
    /** Get connect URI for clients */
    public URI getConnectUri() throws Exception;
    public URI getPublishableConnectURI() throws Exception;
    
    /** Connection statistics */
    public TransportConnectorStatistics getStatistics();
    public void resetStatistics();
}

/**
 * SSL transport connector with security features
 */
public class SslTransportConnector extends TransportConnector {
    /** Set key store configuration */
    public void setKeyStore(String keyStore);
    public String getKeyStore();
    public void setKeyStorePassword(String keyStorePassword);
    
    /** Set trust store configuration */
    public void setTrustStore(String trustStore);
    public String getTrustStore();
    public void setTrustStorePassword(String trustStorePassword);
    
    /** SSL protocol configuration */
    public void setWantClientAuth(boolean wantClientAuth);
    public boolean getWantClientAuth();
    public void setNeedClientAuth(boolean needClientAuth);
    public boolean getNeedClientAuth();
}

/**
 * Base network connector for broker clustering
 */
public abstract class NetworkConnector extends DefaultBrokerService implements Service {
    /** Network connector configuration */
    public void setUri(URI uri);
    public URI getUri();
    public void setDiscoveryUri(URI discoveryUri);
    public URI getDiscoveryUri();
    
    /** Network behavior */
    public void setDuplex(boolean duplex);
    public boolean isDuplex();
    public void setNetworkTTL(int networkTTL);
    public int getNetworkTTL();
    
    /** Message flow control */
    public void setConduitSubscriptions(boolean conduitSubscriptions);
    public boolean isConduitSubscriptions();
    public void setDynamicOnly(boolean dynamicOnly);
    public boolean isDynamicOnly();
    
    /** Bridge configuration */
    public void setBridgeTempDestinations(boolean bridgeTempDestinations);
    public boolean isBridgeTempDestinations();
    public void setDecreaseNetworkConsumerPriority(boolean decreaseNetworkConsumerPriority);
    public boolean isDecreaseNetworkConsumerPriority();
    
    /** Network filters */
    public void setExcludedDestinations(ActiveMQDestination[] excludedDestinations);
    public ActiveMQDestination[] getExcludedDestinations();
    public void setDynamicallyIncludedDestinations(ActiveMQDestination[] dynamicallyIncludedDestinations);
    public ActiveMQDestination[] getDynamicallyIncludedDestinations();
    public void setStaticallyIncludedDestinations(ActiveMQDestination[] staticallyIncludedDestinations);
    public ActiveMQDestination[] getStaticallyIncludedDestinations();
    
    /** Start network connector */
    public void start() throws Exception;
    
    /** Stop network connector */
    public void stop() throws Exception;
}

/**
 * Discovery-based network connector
 */
public class DiscoveryNetworkConnector extends NetworkConnector {
    /** Set discovery agent for finding brokers */
    public void setDiscoveryAgent(DiscoveryAgent discoveryAgent);
    public DiscoveryAgent getDiscoveryAgent();
}

/**
 * Multicast network connector
 */
public class MulticastNetworkConnector extends DiscoveryNetworkConnector {
    /** Multicast configuration */
    public void setMulticastGroup(String multicastGroup);
    public String getMulticastGroup();
    public void setMulticastPort(int multicastPort);
    public int getMulticastPort();
}

/**
 * Interface for broker plugins
 */
public interface BrokerPlugin {
    /** Install plugin into broker */
    Broker installPlugin(Broker broker) throws Exception;
}

/**
 * Statistics collection plugin
 */
public class StatisticsBrokerPlugin implements BrokerPlugin {
    public Broker installPlugin(Broker broker) throws Exception;
}

/**
 * Authentication plugin for security
 */
public class SimpleAuthenticationPlugin implements BrokerPlugin {
    /** Set user/password authentication */
    public void setUsers(Map<String, String> users);
    public Map<String, String> getUsers();
    
    /** Set group memberships */
    public void setGroups(Map<String, Set<String>> groups);
    public Map<String, Set<String>> getGroups();
    
    public Broker installPlugin(Broker broker) throws Exception;
}

/**
 * Authorization plugin for access control
 */
public class AuthorizationPlugin implements BrokerPlugin {
    /** Set authorization map */
    public void setMap(AuthorizationMap map);
    public AuthorizationMap getMap();
    
    public Broker installPlugin(Broker broker) throws Exception;
}

// Broker with authentication
BrokerService broker = new BrokerService();

// Configure authentication plugin
SimpleAuthenticationPlugin authPlugin = new SimpleAuthenticationPlugin();
Map<String, String> users = new HashMap<>();
users.put("admin", "password");
users.put("user", "userpass");
authPlugin.setUsers(users);

// Configure authorization plugin
AuthorizationPlugin authzPlugin = new AuthorizationPlugin();
DefaultAuthorizationMap authMap = new DefaultAuthorizationMap();
// Configure authorization rules...
authzPlugin.setMap(authMap);

// Install plugins
broker.setPlugins(new BrokerPlugin[]{authPlugin, authzPlugin});
broker.start();

/**
 * System usage configuration for resource management
 */
public class SystemUsage {
    /** Memory usage limits */
    public void setMemoryUsage(MemoryUsage memoryUsage);
    public MemoryUsage getMemoryUsage();
    
    /** Store usage limits */
    public void setStoreUsage(StoreUsage storeUsage);
    public StoreUsage getStoreUsage();
    
    /** Temporary usage limits */
    public void setTempUsage(TempUsage tempUsage);
    public TempUsage getTempUsage();
    
    /** Job scheduler usage limits */
    public void setJobSchedulerUsage(JobSchedulerUsage jobSchedulerUsage);
    public JobSchedulerUsage getJobSchedulerUsage();
    
    /** Check if usage exceeds limits */
    public boolean isFull();
    
    /** Wait for usage to drop below limits */
    public void waitForSpace() throws InterruptedException;
    public boolean waitForSpace(long timeout) throws InterruptedException;
}

/**
 * Memory usage tracking and limits
 */
public class MemoryUsage extends Usage<MemoryUsage> {
    /** Set memory limit in bytes */
    public void setLimit(long limit);
    public long getLimit();
    
    /** Get current usage */
    public long getUsage();
    public int getPercentUsage();
    
    /** Check usage status */
    public boolean isFull();
}

/**
 * Store usage tracking and limits
 */
public class StoreUsage extends Usage<StoreUsage> {
    /** Set store size limit */
    public void setLimit(long limit);
    public long getLimit();
    
    /** Get current store usage */
    public long getUsage();
    public int getPercentUsage();
    
    /** Set store directory */
    public void setDirectory(File directory);
    public File getDirectory();
}

// Configure system usage limits
BrokerService broker = new BrokerService();

SystemUsage systemUsage = new SystemUsage();

// Set memory limit to 512MB
MemoryUsage memoryUsage = new MemoryUsage();
memoryUsage.setLimit(512 * 1024 * 1024);
systemUsage.setMemoryUsage(memoryUsage);

// Set store limit to 10GB
StoreUsage storeUsage = new StoreUsage();
storeUsage.setLimit(10L * 1024 * 1024 * 1024);
systemUsage.setStoreUsage(storeUsage);

broker.setSystemUsage(systemUsage);
broker.start();

/**
 * Broker plugin support base class
 */
public class BrokerPluginSupport extends BrokerFilter implements BrokerPlugin {
    public Broker installPlugin(Broker broker) throws Exception;
}

/**
 * Management context for JMX integration
 */
public class ManagementContext {
    /** JMX configuration */
    public void setJmxDomainName(String jmxDomainName);
    public String getJmxDomainName();
    public void setCreateConnector(boolean createConnector);
    public boolean isCreateConnector();
    
    /** Start JMX services */
    public void start() throws IOException;
    
    /** Stop JMX services */
    public void stop() throws IOException;
}

/**
 * Policy map for destination-specific configurations
 */
public class PolicyMap extends BaseDestination {
    /** Set default policy entry */
    public void setDefaultEntry(PolicyEntry defaultEntry);
    public PolicyEntry getDefaultEntry();
    
    /** Set queue-specific policies */
    public void setQueueEntries(List<DestinationMapEntry> queueEntries);
    public List<DestinationMapEntry> getQueueEntries();
    
    /** Set topic-specific policies */
    public void setTopicEntries(List<DestinationMapEntry> topicEntries);
    public List<DestinationMapEntry> getTopicEntries();
}