servi/org.eclipse.statet.rj.servi/srcServiPool/org/eclipse/statet/internal/rj/servi/PoolManager.java

/*=============================================================================#
 # Copyright (c) 2009, 2020 Stephan Wahlbrink and others.
 # 
 # This program and the accompanying materials are made available under the
 # terms of the Eclipse Public License 2.0 which is available at
 # https://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
 # which is available at https://www.apache.org/licenses/LICENSE-2.0.
 # 
 # SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
 # 
 # Contributors:
 #     Stephan Wahlbrink <sw@wahlbrink.eu> - initial API and implementation
 #=============================================================================*/

package org.eclipse.statet.internal.rj.servi;

import static org.eclipse.statet.jcommons.lang.ObjectUtils.nonNullAssert;

import java.rmi.Remote;
import java.rmi.server.RMIClientSocketFactory;
import java.rmi.server.RMIServerSocketFactory;
import java.rmi.server.UnicastRemoteObject;
import java.util.Collection;
import java.util.NoSuchElementException;

import javax.rmi.ssl.SslRMIClientSocketFactory;
import javax.rmi.ssl.SslRMIServerSocketFactory;

import org.eclipse.statet.jcommons.collections.CopyOnWriteIdentityListSet;
import org.eclipse.statet.jcommons.collections.ImCollection;
import org.eclipse.statet.jcommons.collections.ImCollections;
import org.eclipse.statet.jcommons.lang.NonNullByDefault;
import org.eclipse.statet.jcommons.lang.Nullable;
import org.eclipse.statet.jcommons.rmi.RMIRegistry;

import org.eclipse.statet.rj.RjException;
import org.eclipse.statet.rj.RjInitFailedException;
import org.eclipse.statet.rj.server.ServerLogin;
import org.eclipse.statet.rj.servi.RServi;
import org.eclipse.statet.rj.servi.node.RServiNodeFactory;
import org.eclipse.statet.rj.servi.node.RServiPool;
import org.eclipse.statet.rj.servi.pool.PoolConfig;
import org.eclipse.statet.rj.servi.pool.PoolNodeObject;
import org.eclipse.statet.rj.servi.pool.RServiPoolManager;


@NonNullByDefault
public class PoolManager implements RServiPool, RServiPoolManager {
	
	
	private static final int STOP_ON_ERROR= 1 << 24;
	
	
	private final String id;
	
	private final @Nullable RMIRegistry registry;
	
	private @Nullable Remote thisRemote;
	
	private @Nullable APool2 pool;
	private @Nullable APool2NodeFactory poolFactory;
	private PoolConfig poolConfig;
	
	private NodeFactory nodeFactory;
	
	private final CopyOnWriteIdentityListSet<PoolListener> poolListeners= new CopyOnWriteIdentityListSet<>();
	private final Stats stats;
	
	
	public PoolManager(final String id, final @Nullable RMIRegistry registry) {
		this.id= nonNullAssert(id);
		this.registry= registry;
		this.stats= new Stats();
		this.poolListeners.add(this.stats);
		this.poolConfig= new PoolConfig();
		
		Utils.preLoad();
	}
	
	
	@Override
	public String getId() {
		return this.id;
	}
	
	@Override
	public ImCollection<? extends NodeFactory> getFactories() {
		return ImCollections.newList(this.nodeFactory);
	}
	
	@Override
	public synchronized void addNodeFactory(final RServiNodeFactory factory) {
		this.nodeFactory= (NodeFactory) factory;
	}
	
	@Override
	public synchronized void setConfig(final PoolConfig config) {
		if (this.pool != null) {
			this.pool.setConfig(config);
			this.poolFactory.setMaxUsageCount(config.getMaxUsageCount());
		}
		this.poolConfig= config;
	}
	
	@Override
	public PoolConfig getConfig() {
		return this.poolConfig;
	}
	
	public void addPoolListener(final PoolListener listener) {
		this.poolListeners.add(listener);
	}
	
	public void removePoolListener(final PoolListener listener) {
		this.poolListeners.remove(listener);
	}
	
	@Override
	public synchronized void init() throws RjException {
		this.poolFactory= new APool2NodeFactory(this.nodeFactory, this.poolListeners);
		this.poolFactory.setMaxUsageCount(this.poolConfig.getMaxUsageCount());
		this.pool= new APool2(this.poolFactory, this.poolConfig);
		
		Utils.logInfo("Publishing pool in registry...");
		if (this.registry != null) {
			RMIClientSocketFactory clientSocketFactory= null;
			RMIServerSocketFactory serverSocketFactory= null;
			if (this.registry.getAddress().isSsl()) {
				clientSocketFactory= new SslRMIClientSocketFactory();
				serverSocketFactory= new SslRMIServerSocketFactory(null, null, true);
			}
			try {
				final Remote remote= UnicastRemoteObject.exportObject(this, 0,
						clientSocketFactory, serverSocketFactory );
				this.thisRemote= remote;
				this.registry.getRegistry().rebind(PoolConfig.getPoolName(this.id), remote);
			}
			catch (final Exception e) {
				try {
					stop(STOP_ON_ERROR);
				}
				catch (final Exception ignore) {}
				Utils.logError("An error occurred when binding the pool in the registry.", e);
				throw new RjInitFailedException("An error occurred when publishing the pool in the registry.");
			}
		}
	}
	
	@Override
	public Collection<? extends PoolNodeObject> getPoolNodeObjects() {
		if (!isInitialized()) {
			throw new IllegalStateException("not initialized");
		}
		return this.poolFactory.getAllObjects();
	}
	
	public boolean isInitialized() {
		return (this.pool != null);
	}
	
	@Override
	public synchronized void stop(final int mode) throws RjException {
		Utils.logInfo("Unpublishing pool...");
		if (this.registry != null) {
			try {
				this.registry.getRegistry().unbind(PoolConfig.getPoolName(this.id));
			}
			catch (final Exception e) {
				if ((mode & STOP_ON_ERROR) != 0) {
					Utils.logError("An error occurred when unbinding the pool from the registry.", e);
				}
			}
		}
		if (this.thisRemote != null) {
			try {
				this.thisRemote= null;
				UnicastRemoteObject.unexportObject(this, true);
			}
			catch (final Exception e) {
				if ((mode & STOP_ON_ERROR) != 0) {
					Utils.logError("An error occurred when unexport the pool.", e);
				}
			}
		}
		
		try {
			Thread.sleep(500);
		}
		catch (final InterruptedException e) {
		}
		if (this.pool != null) {
			Utils.logInfo("Closing R nodes...");
			try {
				this.pool.close(this.poolConfig.getEvictionTimeout());
			}
			catch (final Exception e) {
				Utils.logError("An error occurred when closing the pool.", e);
			}
			finally {
				Runtime.getRuntime().gc();
			}
		}
	}
	
	@Override
	public RServi getRServi(final String name, final ServerLogin login) throws NoSuchElementException, RjException {
		final APool2NodeHandler handler= getPoolObject(name);
		return new RServiImpl<>(handler.getAccessId(), handler, handler.getClientHandler());
	}
	
	private APool2NodeHandler getPoolObject(final String client) throws NoSuchElementException, RjException {
		try {
			return this.pool.borrowObject(client);
		}
		catch (final NoSuchElementException e) {
			this.stats.logServRequestFailed(3);
			throw new NoSuchElementException(Messages.GetRServi_NoInstance_pub_Pool_message);
		}
		catch (final Exception e) {
			this.stats.logServRequestFailed(4);
			Utils.logError(Messages.BindClient_error_message, e);
			throw new RjException(Messages.GetRServi_pub_error_message);
		}
	}
	
	@Override
	public RServiPoolManager.Counter getCounter() {
		if (!isInitialized()) {
			throw new IllegalStateException("not initialized");
		}
		final RServiPoolManager.Counter counter= new RServiPoolManager.Counter();
		synchronized (this.pool) {
			counter.numIdling= this.pool.getNumIdle();
			counter.numInUse= this.pool.getNumActive();
			counter.numTotal= counter.numIdling + counter.numTotal;
			counter.maxIdling= -1;
			counter.maxInUse= this.poolFactory.getStatMaxAllocated();
			counter.maxTotal= this.poolFactory.getStatMaxTotal();
		}
		return counter;
	}
	
}