/*

  * #%L

  * SocketConnectionFactory.java - mongodb-async-driver - Allanbank Consulting, Inc.

  * %%

  * Copyright (C) 2011 - 2014 Allanbank Consulting, Inc.

  * %%

  * Licensed under the Apache License, Version 2.0 (the "License");

  * you may not use this file except in compliance with the License.

  * You may obtain a copy of the License at

  * 

  *      http://www.apache.org/licenses/LICENSE-2.0

  * 

  * Unless required by applicable law or agreed to in writing, software

  * distributed under the License is distributed on an "AS IS" BASIS,

  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

  * See the License for the specific language governing permissions and

  * limitations under the License.

  * #L%

  */

 package com.allanbank.mongodb.client.connection.socket;

 import java.beans.PropertyChangeEvent;

 import java.beans.PropertyChangeListener;

 import java.io.IOException;

 import java.lang.ref.Reference;

 import java.net.InetSocketAddress;

 import java.util.ArrayList;

 import java.util.Collections;

 import java.util.List;

 import java.util.concurrent.ExecutionException;

 import com.allanbank.mongodb.MongoClientConfiguration;

 import com.allanbank.mongodb.Version;

 import com.allanbank.mongodb.bson.io.BufferingBsonOutputStream;

 import com.allanbank.mongodb.bson.io.StringDecoderCache;

 import com.allanbank.mongodb.bson.io.StringEncoderCache;

 import com.allanbank.mongodb.client.ClusterStats;

 import com.allanbank.mongodb.client.ClusterType;

 import com.allanbank.mongodb.client.connection.Connection;

 import com.allanbank.mongodb.client.connection.ConnectionFactory;

 import com.allanbank.mongodb.client.connection.ReconnectStrategy;

 import com.allanbank.mongodb.client.connection.proxy.ProxiedConnectionFactory;

 import com.allanbank.mongodb.client.message.IsMaster;

 import com.allanbank.mongodb.client.state.Cluster;

 import com.allanbank.mongodb.client.state.LatencyServerSelector;

 import com.allanbank.mongodb.client.state.Server;

 import com.allanbank.mongodb.client.state.ServerSelector;

 import com.allanbank.mongodb.client.state.ServerUpdateCallback;

 import com.allanbank.mongodb.client.state.SimpleReconnectStrategy;

 import com.allanbank.mongodb.util.log.Log;

 import com.allanbank.mongodb.util.log.LogFactory;

 /**

  * {@link ConnectionFactory} to create direct socket connections to the servers.

  * 

  * @api.no This class is <b>NOT</b> part of the drivers API. This class may be

  *         mutated in incompatible ways between any two releases of the driver.

  * @copyright 2011-2014, Allanbank Consulting, Inc., All Rights Reserved

  */

 public class SocketConnectionFactory implements ProxiedConnectionFactory {

     /** The logger for the factory. */

     private static final Log LOG = LogFactory

             .getLog(SocketConnectionFactory.class);

     /**

      * The buffers used by the single threaded connections. Each buffer is

      * shared by all connections but there can be up to 1 buffer per application

      * thread. We use a reference to the buffer to allow the garbage collector

      * To clean up the stream.

      */

     private ThreadLocal<Reference<BufferingBsonOutputStream>> myBuffers;

     /** The state of the cluster. */

     private final Cluster myCluster;

     /** The MongoDB client configuration. */

     private final MongoClientConfiguration myConfig;

     /** The MongoDB client configuration. */

     private final ConfigurationListener myConfigListener;

     /** Cache used for decoding strings. */

     private final StringDecoderCache myDecoderCache;

     /** Cache used for encoding strings. */

     private final StringEncoderCache myEncoderCache;

     /** The server selector. */

     private final ServerSelector myServerSelector;

     /**

      * Creates a new {@link SocketConnectionFactory}.

      * 

      * @param config

      *            The MongoDB client configuration.

      */

     public SocketConnectionFactory(final MongoClientConfiguration config) {

         super();

         myConfig = config;

         myCluster = new Cluster(config, ClusterType.STAND_ALONE);

         myServerSelector = new LatencyServerSelector(myCluster, true);

         myBuffers = new ThreadLocal<Reference<BufferingBsonOutputStream>>();

         myConfigListener = new ConfigurationListener();

         myConfig.addPropertyChangeListener(myConfigListener);

         myDecoderCache = new StringDecoderCache();

         myDecoderCache.setMaxCacheEntries(config.getMaxCachedStringEntries());

         myDecoderCache.setMaxCacheLength(config.getMaxCachedStringLength());

         myEncoderCache = new StringEncoderCache();

         myEncoderCache.setMaxCacheEntries(config.getMaxCachedStringEntries());

         myEncoderCache.setMaxCacheLength(config.getMaxCachedStringLength());

     }

     /**

      * {@inheritDoc}

      * <p>

      * Overridden to do nothing.

      * </p>

      */

     @Override

     public void close() {

         myBuffers = null; // Let the ThreadLocal's weak reference go.

         myConfig.removePropertyChangeListener(myConfigListener);

         // Release the cached entries too.

         myDecoderCache.setMaxCacheEntries(0);

         myDecoderCache.setMaxCacheLength(0);

     }

     /**

      * {@inheritDoc}

      * <p>

      * Returns a new {@link SocketConnection}.

      * </p>

      * 

      * @see ConnectionFactory#connect()

      */

     @Override

     public Connection connect() throws IOException {

         final List<InetSocketAddress> servers = new ArrayList<InetSocketAddress>(

                 myConfig.getServerAddresses());

         // Shuffle the servers and try to connect to each until one works.

         IOException last = null;

         Collections.shuffle(servers);

         for (final InetSocketAddress address : servers) {

             try {

                 final Server server = myCluster.add(address);

                 final Connection conn = connect(server, myConfig);

                 // Get the state of the server updated.

                 final ServerUpdateCallback cb = new ServerUpdateCallback(server);

                 conn.send(new IsMaster(), cb);

                 if (Version.UNKNOWN.equals(server.getVersion())) {

                     // If we don't know the version then wait for that response.

                     try {

                         cb.get();

                     }

                     catch (final ExecutionException e) {

                         // Probably not in a good state...

                         LOG.debug(e, "Could not execute an 'ismaster' command.");

                     }

                     catch (final InterruptedException e) {

                         // Probably not in a good state...

                         LOG.debug(e, "Could not execute an 'ismaster' command.");

                     }

                 }

                 return conn;

             }

             catch (final IOException error) {

                 last = error;

             }

         }

         if (last != null) {

             throw last;

         }

         throw new IOException("Could not connect to any server: " + servers);

     }

     /**

      * Creates a connection to the address provided.

      * 

      * @param server

      *            The MongoDB server to connect to.

      * @param config

      *            The configuration for the Connection to the MongoDB server.

      * @return The Connection to MongoDB.

      * @throws IOException

      *             On a failure connecting to the server.

      */

     @Override

     public Connection connect(final Server server,

             final MongoClientConfiguration config) throws IOException {

         final AbstractSocketConnection connection;

         switch (myConfig.getConnectionModel()) {

         case SENDER_RECEIVER_THREAD: {

             connection = new TwoThreadSocketConnection(server, myConfig,

                     myEncoderCache, myDecoderCache);

             break;

         }

         default: { // and RECEIVER_THREAD

             connection = new SocketConnection(server, myConfig, myEncoderCache,

                     myDecoderCache, myBuffers);

             break;

         }

         }

         // Start the connection.

         connection.start();

         return connection;

     }

     /**

      * Returns the cluster state.

      * 

      * @return The cluster state.

      */

     public Cluster getCluster() {

         return myCluster;

     }

     /**

      * {@inheritDoc}

      * <p>

      * Overridden to return the {@link Cluster}.

      * </p>

      */

     @Override

     public ClusterStats getClusterStats() {

         return myCluster;

     }

     /**

      * {@inheritDoc}

      * <p>

      * Overridden to return {@link ClusterType#STAND_ALONE} cluster type.

      * </p>

      */

     @Override

     public ClusterType getClusterType() {

         return ClusterType.STAND_ALONE;

     }

     /**

      * {@inheritDoc}

      * <p>

      * Overridden to return a {@link SimpleReconnectStrategy}.

      * </p>

      */

     @Override

     public ReconnectStrategy getReconnectStrategy() {

         final SimpleReconnectStrategy strategy = new SimpleReconnectStrategy();

         strategy.setConfig(myConfig);

         strategy.setConnectionFactory(this);

         strategy.setSelector(myServerSelector);

         strategy.setState(myCluster);

         return strategy;

     }

     /**

      * Notification of a change to the configuration of the client.

      * 

      * @param evt

      *            The details of the configuration change.

      */

     protected void configurationChanged(final PropertyChangeEvent evt) {

         final String name = evt.getPropertyName();

         final Object value = evt.getNewValue();

         if ("maxCachedStringEntries".equals(name) && (value instanceof Number)) {

             myDecoderCache.setMaxCacheEntries(((Number) value).intValue());

             myEncoderCache.setMaxCacheEntries(((Number) value).intValue());

         }

         else if ("maxCachedStringLength".equals(name)

                 && (value instanceof Number)) {

             myDecoderCache.setMaxCacheLength(((Number) value).intValue());

             myEncoderCache.setMaxCacheLength(((Number) value).intValue());

         }

     }

     /**

      * ConfigurationListener provides a listener for changes in the client's

      * configuration.

      * 

      * @api.no This class is <b>NOT</b> part of the drivers API. This class may

      *         be mutated in incompatible ways between any two releases of the

      *         driver.

      * @copyright 2014, Allanbank Consulting, Inc., All Rights Reserved

      */

     protected final class ConfigurationListener implements

             PropertyChangeListener {

         @Override

         public void propertyChange(final PropertyChangeEvent evt) {

             configurationChanged(evt);

         }

     }

 }