/*
 * JBoss, Home of Professional Open Source.
 * See the COPYRIGHT.txt file distributed with this work for information
 * regarding copyright ownership.  Some portions may be licensed
 * to Red Hat, Inc. under one or more contributor license agreements.
 * 
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 * 
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
 * 02110-1301 USA.
 * Modified by
 * @author  Data Virtuality GmbH
 * @date    $Date$
 *  
 */
/*
 * EXTENSION
 * ISSUE: DVCORE-2521
 * FIXED BY: Roman Pedchenko
 * DATE: 2014-12-05
 * ADOPTED FROM: TEIID-2153, 2171, 2359, 2149, 2358, 2410, 2271, 2467, 2429, 2363, 2160, 2461, 2797, 2326, 1557
 * END EXTENSION
 */
// https://issues.jboss.org/browse/TEIID-247, 2166, 2151, 2234, 2581, 2571, 2369, 1386, 2226, 2233, 2224, 2225, 2257
// https://issues.jboss.org/browse/TEIID-2326
// https://issues.jboss.org/browse/TEIID-1386
// https://issues.jboss.org/browse/TEIID-2226
// https://issues.jboss.org/browse/TEIID-2168
// https://issues.jboss.org/browse/TEIID-2205
// https://issues.jboss.org/browse/TEIID-2148
// https://issues.jboss.org/browse/TEIID-2423 (partially, only the first commit)
// https://issues.jboss.org/browse/TEIID-2601
// https://issues.jboss.org/browse/TEIID-2328 (partially, only changes for the class)
// https://issues.jboss.org/browse/TEIID-2444
// https://issues.jboss.org/browse/TEIID-2215
// https://issues.jboss.org/browse/TEIID-3519
// https://issues.redhat.com/browse/TEIID-3658
// https://issues.redhat.com/browse/TEIID-2195
// https://issues.redhat.com/browse/TEIID-2664
// https://issues.redhat.com/browse/TEIID-2439
// https://issues.redhat.com/browse/TEIID-2577
// https://issues.redhat.com/browse/TEIID-2116
// https://issues.redhat.com/browse/TEIID-2285
// https://issues.redhat.com/browse/TEIID-3393
// https://issues.redhat.com/browse/TEIID-2172
// https://issues.redhat.com/browse/TEIID-5245
// https://issues.redhat.com/browse/TEIID-4535
package org.teiid.runtime;


import java.io.InputStream;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.InetSocketAddress;
import java.sql.CallableStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

import javax.transaction.RollbackException;
import javax.transaction.Synchronization;
import javax.transaction.SystemException;
import javax.transaction.Transaction;
import javax.transaction.TransactionManager;
import javax.xml.stream.XMLStreamException;

import org.teiid.adminapi.Admin;
import org.teiid.adminapi.impl.ModelMetaData;
import org.teiid.adminapi.impl.VDBMetaData;
import org.teiid.adminapi.impl.VDBMetadataParser;
import org.teiid.client.DQP;
import org.teiid.client.security.ILogon;
import org.teiid.common.buffer.BufferManager;
import org.teiid.common.buffer.TupleBufferCache;
import org.teiid.core.TeiidRuntimeException;
import org.teiid.core.BundleUtil.Event;
import org.teiid.core.TeiidException;
import org.teiid.deployers.*;
import org.teiid.dqp.internal.datamgr.ConnectorManager;
import org.teiid.dqp.internal.datamgr.ConnectorManagerRepository;
import org.teiid.dqp.internal.datamgr.ConnectorManagerRepository.ConnectorManagerException;
import org.teiid.dqp.internal.datamgr.ConnectorManagerRepository.ExecutionFactoryProvider;
import org.teiid.dqp.internal.process.*;
import org.teiid.dqp.service.BufferService;
import org.teiid.dqp.service.TransactionContext;
import org.teiid.dqp.service.TransactionContext.Scope;
import org.teiid.events.EventDistributor;
import org.teiid.events.EventDistributorFactory;
import org.teiid.jdbc.*;
import org.teiid.logging.LogConstants;
import org.teiid.logging.LogManager;
import org.teiid.logging.MessageLevel;
import org.teiid.metadata.MetadataFactory;
import org.teiid.metadata.MetadataRepository;
import org.teiid.metadata.MetadataStore;
import org.teiid.net.ConnectionException;
import org.teiid.net.ServerConnection;
import org.teiid.net.socket.ObjectChannel;
import org.teiid.query.ObjectReplicator;
import org.teiid.query.function.SystemFunctionManager;
import org.teiid.query.metadata.TransformationMetadata;
import org.teiid.query.metadata.TransformationMetadata.Resource;
import org.teiid.query.sql.lang.Command;
import org.teiid.query.tempdata.GlobalTableStore;
import org.teiid.query.validator.ValidatorFailure;
import org.teiid.query.validator.ValidatorReport;
import org.teiid.services.AbstractEventDistributorFactoryService;
import org.teiid.services.BufferServiceImpl;
import org.teiid.services.SessionServiceImpl;
import org.teiid.translator.ExecutionFactory;
import org.teiid.translator.Translator;
import org.teiid.translator.TranslatorException;
import org.teiid.transport.*;
import org.teiid.vdb.runtime.VDBKey;

/**
 * A simplified server environment for embedded use.
 * 
 * Needs to be started prior to use with a call to {@link #start(EmbeddedConfiguration)}
 */
@SuppressWarnings("serial")
public class EmbeddedServer extends AbstractVDBDeployer implements EventDistributorFactory, ExecutionFactoryProvider {

	private EmbeddedProfile embeddedProfile = new EmbeddedProfile() {
		@Override
		public ConnectionImpl connect(String url, Properties info)
				throws TeiidSQLException {
			ServerConnection conn;
			try {
				conn = createServerConnection(info);
			} catch (TeiidException e) {
				throw TeiidSQLException.create(e);
			}
			return new EmbeddedConnectionImpl(conn, info, url);
		}

		@Override
		public ServerConnection createServerConnection(Properties info)
				throws TeiidException {
			LocalServerConnection conn = new LocalServerConnection(info, useCallingThread) {
				@Override
				protected ClientServiceRegistry getClientServiceRegistry(String name) {
					return services;
				}
			};
			conn.getWorkContext().setConnectionProfile(this);
			return conn;
		}
	};

	private EmbeddedConfiguration config;

	private final class EmbeddedConnectionImpl extends ConnectionImpl implements EmbeddedConnection {

		public EmbeddedConnectionImpl(ServerConnection serverConn,
				Properties info, String url) {
			super(serverConn, info, url);
		}

		@Override
		public CallableStatement prepareCall(Command command, EmbeddedRequestOptions options)
				throws SQLException {
			CallableStatementImpl csi = this.prepareCall(command.toString(), options.getResultSetType(), ResultSet.CONCUR_READ_ONLY);
			csi.setCommand(command);
			return csi;
		}
		
		@Override
		public TeiidPreparedStatement prepareStatement(Command command,
				EmbeddedRequestOptions options) throws SQLException {
			PreparedStatementImpl psi = this.prepareStatement(command.toString(), options.getResultSetType(), ResultSet.CONCUR_READ_ONLY);
			psi.setCommand(command);
			return psi;
		}
		
	}

	protected final class TransactionDetectingTransactionServer extends
			TransactionServerImpl {
		
		/**
		 * Override to detect existing thread bound transactions.
		 * This may be of interest for local connections as well, but
		 * we assume there that a managed datasource will be used.
		 * A managed datasource is not possible here.
		 */
		public TransactionContext getOrCreateTransactionContext(final String threadId) {
			TransactionContext tc = super.getOrCreateTransactionContext(threadId);
			if (useCallingThread && detectTransactions && tc.getTransaction() == null) {
				try {
					Transaction tx = transactionManager.getTransaction();
					if (tx != null) {
						tx.registerSynchronization(new Synchronization() {
							
							@Override
							public void beforeCompletion() {
							}
							
							@Override
							public void afterCompletion(int status) {
								transactions.removeTransactionContext(threadId);
							}
						});
						tc.setTransaction(tx);
						tc.setTransactionType(Scope.GLOBAL);
					}
				} catch (SystemException e) {
				} catch (IllegalStateException e) {
				} catch (RollbackException e) {
				}
			}
			
			return tc;
		}
	}

	protected class ProviderAwareConnectorManagerRepository extends
			ConnectorManagerRepository {

		/*public ProviderAwareConnectorManagerRepository() {
			super(true);
		}*/

		@Override
		protected ConnectorManager createConnectorManager(
				String translatorName, String connectionName, ExecutionFactory<Object, Object> ef) throws ConnectorManagerException {
			return new ConnectorManager(translatorName, connectionName, ef) {
				@Override
				public Object getConnectionFactory() throws TranslatorException {
					if (getConnectionName() == null) {
						return null;
					}
					ConnectionFactoryProvider<?> connectionFactoryProvider = connectionFactoryProviders.get(getConnectionName());
					if (connectionFactoryProvider != null) {
						return connectionFactoryProvider.getConnectionFactory();
					}
					return super.getConnectionFactory();
				}
			};
		}
	}

	public interface ConnectionFactoryProvider<T> {
		T getConnectionFactory() throws TranslatorException;
	}

	public static class SimpleConnectionFactoryProvider<T> implements ConnectionFactoryProvider<T> {

		private T connectionFactory;

		public SimpleConnectionFactoryProvider(T connectionFactory) {
			this.connectionFactory = connectionFactory;
		}

		@Override
		public T getConnectionFactory() throws TranslatorException {
			return connectionFactory;
		}

	}

	private static class VDBValidationError extends TeiidRuntimeException {
		
		private VDBValidationError(Event event, String message) {
			super(event, message);
		}
	}
	
	protected DQPCore dqp = new DQPCore();
	/**
	 * Custom vdb repository that will immediately throw exceptions for metadata validation errors
	 */
	protected VDBRepository repo = new VDBRepository() {
		@Override
		protected boolean processMetadataValidatorReport(VDBKey key, ValidatorReport report) {
			if (throwMetadataErrors) {
				super.processMetadataValidatorReport(key, report); //remove
				ValidatorFailure firstFailure = report.getItems().iterator().next();
				throw new VDBValidationError(RuntimePlugin.Event.TEIID40095, firstFailure.getMessage());
			}
			return true;
		}
	};

	class EmbeddedEventDistributorFactoryService extends AbstractEventDistributorFactoryService {

		@Override
 	    protected VDBRepository getVdbRepository() {
			return repo;
		}

		@Override
 	    protected ObjectReplicator getObjectReplicator() {
			return replicator;
		}
 	};

	protected boolean throwMetadataErrors = true;
	private ConcurrentHashMap<String, ExecutionFactory<?, ?>> translators = new ConcurrentHashMap<String, ExecutionFactory<?, ?>>();
	private ConcurrentHashMap<String, ConnectionFactoryProvider<?>> connectionFactoryProviders = new ConcurrentHashMap<String, ConnectionFactoryProvider<?>>();
	protected SessionServiceImpl sessionService = new SessionServiceImpl();
	protected ObjectReplicator replicator;
	protected BufferServiceImpl bufferService = new BufferServiceImpl();
	protected TransactionDetectingTransactionServer transactionService = new TransactionDetectingTransactionServer();
	protected boolean waitForLoad;
	protected ClientServiceRegistryImpl services = new ClientServiceRegistryImpl() {
		public void waitForFinished(String vdbName, int vdbVersion, int timeOutMillis) throws ConnectionException {
			if (waitForLoad) {
				repo.waitForFinished(vdbName, vdbVersion, timeOutMillis);
			}
		}
		@Override
		public ClassLoader getCallerClassloader() {
			return this.getClass().getClassLoader();
		};
	};
	protected LogonImpl logon;
	private TeiidDriver driver = new TeiidDriver();
	protected ConnectorManagerRepository cmr = new ProviderAwareConnectorManagerRepository();

	protected AbstractEventDistributorFactoryService eventDistributorFactoryService;
	protected boolean useCallingThread = true;
	//TODO: allow for configurablity - in environments that support subtransations it would be fine
	//to allow teiid to start a request transaction under an existing thread bound transaction
	protected boolean detectTransactions = true;
	private Boolean running;
	private SessionAwareCache<CachedResults> rs;
	private SessionAwareCache<PreparedPlan> ppc;
	protected ArrayList<SocketListener> transports = new ArrayList<SocketListener>();

	public EmbeddedServer() {

	}

	public void addConnectionFactoryProvider(String name,
			ConnectionFactoryProvider<?> connectionFactoryProvider) {
		this.connectionFactoryProviders.put(name, connectionFactoryProvider);
	}

	/**
	 * Adds the object as the named connection factory to replace the default JNDI lookup strategy.
	 * @param name
	 * @param connectionFactoryr
	 */
	public void addConnectionFactory(String name, Object connectionFactory) {
		this.connectionFactoryProviders.put(name, new SimpleConnectionFactoryProvider<Object>(connectionFactory));
	}

	public synchronized void start(@SuppressWarnings("hiding") EmbeddedConfiguration config) {
		if (running != null) {
			throw new IllegalStateException();
		}
		this.config = config;
		this.replicator = config.getObjectReplicator();
		this.eventDistributorFactoryService = new EmbeddedEventDistributorFactoryService();
		//must be called after the replicator is set
		this.eventDistributorFactoryService.start();
		this.dqp.setEventDistributor(this.eventDistributorFactoryService.getReplicatedEventDistributor());
		if (config.getTransactionManager() == null) {
			LogManager.logInfo(LogConstants.CTX_RUNTIME, RuntimePlugin.Util.gs(RuntimePlugin.Event.TEIID40089));
			this.transactionService.setTransactionManager((TransactionManager) Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class<?>[] {TransactionManager.class}, new InvocationHandler() {
				
				@Override
				public Object invoke(Object proxy, Method method, Object[] args)
						throws Throwable {
					throw new UnsupportedOperationException(RuntimePlugin.Util.gs(RuntimePlugin.Event.TEIID40089));
				}
			}));
			this.detectTransactions = false;
		} else {
			this.transactionService.setTransactionManager(config.getTransactionManager());
		}
		if (config.getSecurityHelper() != null) {
			this.sessionService.setSecurityHelper(config.getSecurityHelper());
		} else {
			this.sessionService.setSecurityHelper(new DoNothingSecurityHelper());
		}
		if (config.getSecurityDomain() != null) {
			this.sessionService.setSecurityDomain(config.getSecurityDomain());
		} else {
			this.sessionService.setSecurityDomain("dv-security"); //$NON-NLS-1$
		}

		this.sessionService.setVDBRepository(repo);
		this.bufferService.setUseDisk(config.isUseDisk());
		if (config.isUseDisk()) {
			if (config.getBufferDirectory() == null) {
				config.setBufferDirectory(System.getProperty("java.io.tmpdir")); //$NON-NLS-1$
			}
			this.bufferService.setDiskDirectory(config.getBufferDirectory());
		}
		BufferService bs = getBufferService();
		this.dqp.setBufferManager(bs.getBufferManager());

		startVDBRepository();

		rs = new SessionAwareCache<CachedResults>("resultset", config.getCacheFactory(), SessionAwareCache.Type.RESULTSET, config.getMaxResultSetCacheStaleness()); //$NON-NLS-1$
		ppc = new SessionAwareCache<PreparedPlan>("preparedplan", config.getCacheFactory(), SessionAwareCache.Type.PREPAREDPLAN, 0); //$NON-NLS-1$
		rs.setTupleBufferCache(bs.getTupleBufferCache());
		this.dqp.setResultsetCache(rs);

		ppc.setTupleBufferCache(bs.getTupleBufferCache());
		this.dqp.setPreparedPlanCache(ppc);

		this.dqp.setTransactionService(this.transactionService);

		this.dqp.start(config);
		this.sessionService.setDqp(this.dqp);
		this.services.setSecurityHelper(this.sessionService.getSecurityHelper());
		this.logon = new LogonImpl(sessionService, null);
		services.registerClientService(ILogon.class, logon, LogConstants.CTX_SECURITY);
		DQP dqpProxy = DQP.class.cast(Proxy.newProxyInstance(this.getClass().getClassLoader(), new Class[] {DQP.class}, new SessionCheckingProxy(dqp, LogConstants.CTX_DQP, MessageLevel.TRACE)));
		services.registerClientService(DQP.class, dqpProxy, LogConstants.CTX_DQP);
		initDriver();
		List<SocketConfiguration> transports = config.getTransports();
		if ( transports != null && !transports.isEmpty()) {
			for (SocketConfiguration socketConfig:transports) {
				SocketListener socketConnection = startTransport(socketConfig, bs.getBufferManager(), config.getMaxODBCLobSizeAllowed());
				if (socketConnection != null) {
					this.transports.add(socketConnection);
				}
			}
		}
		running = true;
	}

	private void initDriver() {
		driver.setEmbeddedProfile(embeddedProfile);
	}

	private SocketListener startTransport(SocketConfiguration socketConfig, BufferManager bm, int maxODBCLobSize) {
		InetSocketAddress address = new InetSocketAddress(socketConfig.getHostAddress(), socketConfig.getPortNumber());
		if (socketConfig.getProtocol() == WireProtocol.teiid) {
			return new SocketListener(address, socketConfig, this.services, bm) {
				@Override
				public ChannelListener createChannelListener(ObjectChannel channel) {
					//TODO: this is a little dirty, but allows us to inject the appropriate connection profile
					SocketClientInstance instance = (SocketClientInstance) super.createChannelListener(channel);
					instance.getWorkContext().setConnectionProfile(embeddedProfile);
					return instance;
				}
			};
		}
		else if (socketConfig.getProtocol() == WireProtocol.pg) {
			ODBCSocketListener odbc = new ODBCSocketListener(address, socketConfig, this.services, bm, maxODBCLobSize, this.logon, driver);
//			odbc.setAuthenticationType(this.sessionService.getAuthenticationType());
			return odbc;
		}
		return null;
	}

	private void startVDBRepository() {
		this.repo.addListener(new VDBLifeCycleListener() {

			@Override
			public void added(String name, int version, CompositeVDB vdb) {

			}

			@Override
			public void removed(String name, int version, CompositeVDB vdb) {
				if (replicator != null) {
					replicator.stop(vdb.getVDB().getAttachment(GlobalTableStore.class));
				}
				rs.clearForVDB(name, 1);
				ppc.clearForVDB(name, 1);
			}

			@Override
			public void finishedDeployment(String name, int version,
					CompositeVDB vdb) {
				GlobalTableStore gts = CompositeGlobalTableStore.createInstance(vdb, dqp.getBufferManager(), replicator);

				vdb.getVDB().addAttchment(GlobalTableStore.class, gts);
			}
		});
		this.repo.setSystemFunctionManager(new SystemFunctionManager());
		this.repo.start();
	}

	protected BufferService getBufferService() {
		bufferService.start();
		if (replicator != null) {
			try {
				final TupleBufferCache tbc = replicator.replicate("$BM$", TupleBufferCache.class, bufferService.getBufferManager(), 0); //$NON-NLS-1$
				return new BufferService() {

					@Override
					public BufferManager getBufferManager() {
						return bufferService.getBufferManager();
					}

					@Override
					public TupleBufferCache getTupleBufferCache() {
						return tbc;
					}

				};
			} catch (Exception e) {
				throw new TeiidRuntimeException(e);
			}
		}
		return bufferService;
	}

	/**
	 * Adds a default instance of the {@link ExecutionFactory} using the default name either from the {@link Translator} annotation or the class name.
	 * @param ef
	 * @throws TranslatorException
	 */
	public void addTranslator(Class<? extends ExecutionFactory> clazz) throws TranslatorException {
		Translator t = clazz.getAnnotation(Translator.class);
		String name = clazz.getName();
		if (t != null) {
			name = t.name();
		}
		try {
			ExecutionFactory<?, ?> instance = clazz.newInstance();
			instance.start();
			addTranslator(name, instance);
		} catch (InstantiationException e) {
			throw new TeiidRuntimeException(e);
		} catch (IllegalAccessException e) {
			throw new TeiidRuntimeException(e);
		}
	}

	/**
	 * Add an {@link ExecutionFactory}.  The {@link ExecutionFactory#start()} method
	 * should have already been called.
	 * @param ef
	 */
	public void addTranslator(ExecutionFactory<?, ?> ef) {
		Translator t = ef.getClass().getAnnotation(Translator.class);
		String name = ef.getClass().getName();
		if (t != null) {
			name = t.name();
		}
		addTranslator(name, ef);
	}
	
	public void addTranslator(String name, ExecutionFactory<?, ?> ef) {
		translators.put(name, ef);
	}

	/**
	 * Deploy the given set of models as vdb name.1
	 * @param name
	 * @param models
	 * @throws ConnectorManagerException
	 * @throws VirtualDatabaseException
	 * @throws TranslatorException
	 */
	public void deployVDB(String name, ModelMetaData... models)
			throws ConnectorManagerException, VirtualDatabaseException, TranslatorException {
		VDBMetaData vdb = new VDBMetaData();
		vdb.setDynamic(true);
		vdb.setName(name);
		vdb.setModels(Arrays.asList(models));
		deployVDB(vdb);
	}

	/**
	 * Deploy a vdb.xml file.  The name and version will be derived from the xml.
	 * @param is which will be closed by this deployment
	 * @throws XMLStreamException
	 * @throws TranslatorException
	 * @throws ConnectorManagerException
	 * @throws VirtualDatabaseException
	 */
	public void deployVDB(InputStream is) throws XMLStreamException, VirtualDatabaseException, ConnectorManagerException, TranslatorException {
		VDBMetaData metadata = VDBMetadataParser.unmarshell(is);
		metadata.setDynamic(true); // TODO remove the line after backporting TEIID-2449 issue
		deployVDB(metadata);
	}

	protected boolean allowOverrideTranslators() {
		return false;
	}

	public void deployVDB(VDBMetaData vdb)
 	                throws ConnectorManagerException, VirtualDatabaseException, TranslatorException {
		checkStarted();
		if (!vdb.getOverrideTranslators().isEmpty() && !allowOverrideTranslators()) {
			throw new VirtualDatabaseException(RuntimePlugin.Event.TEIID40106, RuntimePlugin.Util.gs(RuntimePlugin.Event.TEIID40106, vdb.getName()));
		}
		cmr.createConnectorManagers(vdb, this);
		MetadataStore metadataStore = new MetadataStore();
		UDFMetaData udfMetaData = new UDFMetaData();
		udfMetaData.setFunctionClassLoader(Thread.currentThread().getContextClassLoader());
		this.assignMetadataRepositories(vdb, null);
		repo.addVDB(vdb, metadataStore, new LinkedHashMap<String, Resource>(), udfMetaData, cmr);
		try {
			this.loadMetadata(vdb, cmr, metadataStore);
		} catch (VDBValidationError e) {
			throw new VirtualDatabaseException(RuntimePlugin.Event.valueOf(e.getCode()), e.getMessage());
		}
	}
	
	/**
	 * TODO: consolidate this logic more into the abstract deployer
	 */
	@Override
	protected void loadMetadata(VDBMetaData vdb, ModelMetaData model,
			ConnectorManagerRepository cmr,
			MetadataRepository metadataRepository, MetadataStore store,
			AtomicInteger loadCount) throws TranslatorException {
		MetadataFactory factory = createMetadataFactory(vdb, model);

		ExecutionFactory ef = null;
		Object cf = null;
		
		try {
			ConnectorManager cm = getConnectorManager(model, cmr);
			if (cm != null) {
				ef = cm.getExecutionFactory();
				cf = cm.getConnectionFactory();
			}
		} catch (TranslatorException e) {
			//cf not available
		}
		
		metadataRepository.loadMetadata(factory, ef, cf);		
		metadataLoaded(vdb, model, store, loadCount, factory, true);
	}
	
	public void undeployVDB(String vdbName) {
		checkStarted();
		VDBMetaData removed = this.repo.removeVDB(vdbName, 1);
		if (removed != null)
			removed.getModelMetaDatas().values().stream()
					.flatMap(model -> model.getSourceMappings().stream())
					.forEach(source -> this.cmr.removeConnectorManager(source.getName()));
	}

	/**
	 * Stops the server.  Once stopped it cannot be restarted.
	 */
	public synchronized void stop() {
		if (config != null) {
			config.stop();
		}
		for (SocketListener socket:this.transports) {
			socket.stop();
		}
		this.transports.clear();
		dqp.stop();
		eventDistributorFactoryService.stop();
		config.getCacheFactory().destroy();
		config.setCacheFactory(null);
		bufferService = null;
		dqp = null;
		running = false;
	}

	private synchronized void checkStarted() {
		if (running == null || !running) {
			throw new IllegalStateException();
		}
	}

	public TeiidDriver getDriver() {
		checkStarted();
		return driver;
	}
	
	@Override
	public EventDistributor getEventDistributor() {
		return this.eventDistributorFactoryService.getEventDistributor();
	}

	@SuppressWarnings("unchecked")
	@Override
	public ExecutionFactory<Object, Object> getExecutionFactory(String name)
			throws ConnectorManagerException {
		ExecutionFactory<?, ?> ef = translators.get(name);
		if (ef == null) {
			//TODO: consolidate this exception into the connectormanagerrepository
			throw new ConnectorManagerException(name);
		}
		return (ExecutionFactory<Object, Object>) ef;
	}
	
	@Override
	protected VDBRepository getVDBRepository() {
		return this.repo;
	}

	/**
	 * Return the bound port for the transport number
	 * @param transport
	 * @return
	 */
	public int getPort(int transport) {
		return this.transports.get(transport).getPort();
	}

	protected ConcurrentHashMap<String, ExecutionFactory<?, ?>> getTranslators() {
		return this.translators;
	}

	protected SessionAwareCache<CachedResults> getRsCache() {
		return this.rs;
	}

	protected SessionAwareCache<PreparedPlan> getPpcCache() {
		return this.ppc;
	}

	public Admin getAdmin() {
		return EmbeddedAdminFactory.getInstance().createAdmin(this);
	}

	public DQPCore getDqp() {
		return dqp;
	}
}
