Coherence, managing multiple Extend connections

Coherence*Extend is a protocol which is used for non-members of cluster to get access to Coherence services. Extend is using TCP connection to one of cluster members (which should host proxy service) and use this member as a relay. Normally client process is creating single TCP connection. This connection is shared between allNamedCache instances and threads of client process. To be more precise, it is creating single TCP connection per remote service, but normal you have just one remote cache service (and in some case another remote invocation service).

Of cause TCP connection will failover automatically is case of proxy or network problems, so for most practical cases it is ok to share single TCP connection per process. Proxy member of cluster is acting as a relay (in most cases it doesn’t even deserialize data passing through), so single client process is unlikely to overload proxy process … unless you are using invocation services. In case of invocation service proxy is performing logic on behalf of client and it can be arbitrary complex, so it may be desirable to spread requests across multiple proxy processes.

Here is simple trick, we should create as many remote services as connections we want to have. There is a slight problem, you cannot have same cache name associated with different remote services at the same time … unless you are using multiple cache factories.

Below is snippet of code which is manages creating cache factories and mangling service names to create separate set of Extend connections perExtendConnection instance.

public class ExtendConnection {

    private static Logger LOGGER = Logger.getLogger(ExtendConnection.class.getName());
    
    private static AtomicInteger CONNECTION_COUNTER = new AtomicInteger();

    private int connectionId = CONNECTION_COUNTER.incrementAndGet();
    private ConfigurableCacheFactory cacheFactory;
    private ConcurrentMap<Service, Service> activeServices = new ConcurrentHashMap<Service, Service>(4, 0.5f, 2);
    
    public ExtendConnection(String configFile) {
        cacheFactory = initPrivateCacheFactory(configFile);
    }

    private DefaultConfigurableCacheFactory initPrivateCacheFactory(String configFile) {
        LOGGER.info("New Extend connection #" + connectionId + " is going to be created, config: " + configFile);

        XmlElement xml = XmlHelper.loadFileOrResource(configFile, "Coherence cache configuration for Extend connection #" + connectionId);
        // transforming configuration
        XmlElement schemes = xml.getSafeElement("caching-schemes");
        for(Object o: schemes.getElementList()) {
            XmlElement scheme = (XmlElement) o;
            if (isRemoteScheme(scheme)) {
                String name = scheme.getSafeElement("service-name").getString();
                if (name != null) {
                    String nname = name + "-" + connectionId;
                    scheme.getElement("service-name").setString(nname);
                }
            }
        }
        
        DefaultConfigurableCacheFactory factory = new DefaultConfigurableCacheFactory(xml);
        return factory;
    }
    
    
    private boolean isRemoteScheme(XmlElement scheme) {
        String name = scheme.getName();
        return "remote-cache-scheme".equals(name) || "remote-invocation-scheme".equals(name);
    }


    public NamedCache getCache(String name) {
        NamedCache cache = cacheFactory.ensureCache(name, null);
        Service service = cache.getCacheService();
        activeServices.putIfAbsent(service, service);
        return cache;
    }

    public InvocationService getInvocationService(String serviceName) {
        InvocationService service = (InvocationService) cacheFactory.ensureService(serviceName + "-" + connectionId);
        activeServices.putIfAbsent(service, service);
        return service;
    }

    /**
     * Warning: this method is not concurrency safe, you may get to trouble if you are accessing caches of services via this connection during shutdown.
     */
    public void disconnect() {
        for(Service service:  new ArrayList<Service>(activeServices.keySet())) {
            try {
                if (service.isRunning()) {
                    service.stop();
                }
            }
            catch(Exception e) {
                LOGGER.log(Level.WARNING, "Exception during remote service shutdown", e);
            }
        }
    }
}

Each instance of class above manages physical TCP Extend connection (or multiple if you have multiple remote services in configuration). To create multiple connections just create multiple instances, but make sure that you are not leaking them. Extend connections will not be closed automatically by GC, so you should pool them carefully.

This technique is also useful if you want to keep connections to several different clusters at the same time.