/*

  * #%L

  * SocketConnection.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.io.IOException;

 import java.lang.ref.Reference;

 import java.lang.ref.SoftReference;

 import java.net.Socket;

 import java.net.SocketException;

 import com.allanbank.mongodb.MongoClientConfiguration;

 import com.allanbank.mongodb.MongoDbException;

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

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

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

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

 import com.allanbank.mongodb.client.Message;

 import com.allanbank.mongodb.client.callback.AddressAware;

 import com.allanbank.mongodb.client.callback.ReplyCallback;

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

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

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

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

 import com.allanbank.mongodb.util.IOUtils;

 /**

  * Provides a blocking Socket based connection to a MongoDB server.

  * <p>

  * This version uses single receive thread. Sending of messages is done by the

  * application thread sending the message.

  * </p>

  * <p>

  * This implementation does not perform as well as the

  * {@link TwoThreadSocketConnection} when the {@link Socket} implementation does

  * not have built in buffering of messages or requires acknowledgment of

  * messages before releasing a write request. For that reason this

  * implementation is only used with the normal Java Socket implementations.

  * </p>

  * 

  * @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 SocketConnection extends AbstractSocketConnection {

     /**

      * The buffers used each connection. Each buffer is shared by all

      * connections but there can be up to 1 buffer per application thread.

      */

     private final ThreadLocal<Reference<BufferingBsonOutputStream>> myBuffers;

     /** The thread receiving replies. */

     private final Thread myReceiver;

     /** The sequence for serializing sends. */

     private final Sequence mySendSequence;

     /**

      * Creates a new SocketConnection to a MongoDB server.

      * 

      * @param server

      *            The MongoDB server to connect to.

      * @param config

      *            The configuration for the Connection to the MongoDB server.

      * @throws SocketException

      *             On a failure connecting to the MongoDB server.

      * @throws IOException

      *             On a failure to read or write data to the MongoDB server.

      */

     public SocketConnection(final Server server,

             final MongoClientConfiguration config) throws SocketException,

             IOException {

         this(server, config, new StringEncoderCache(),

                 new StringDecoderCache(),

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

     }

     /**

      * Creates a new SocketConnection to a MongoDB server.

      * 

      * @param server

      *            The MongoDB server to connect to.

      * @param config

      *            The configuration for the Connection to the MongoDB server.

      * @param encoderCache

      *            Cache used for encoding strings.

      * @param decoderCache

      *            Cache used for decoding strings.

      * @param buffers

      *            The buffers used each connection. Each buffer is shared by all

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

      *            thread.

      * @throws SocketException

      *             On a failure connecting to the MongoDB server.

      * @throws IOException

      *             On a failure to read or write data to the MongoDB server.

      */

     public SocketConnection(final Server server,

             final MongoClientConfiguration config,

             final StringEncoderCache encoderCache,

             final StringDecoderCache decoderCache,

             final ThreadLocal<Reference<BufferingBsonOutputStream>> buffers)

             throws SocketException, IOException {

         super(server, config, encoderCache, decoderCache);

         myBuffers = buffers;

         mySendSequence = new Sequence(1L, config.getLockType());

         myReceiver = config.getThreadFactory().newThread(

                 new ReceiveRunnable(this));

         myReceiver.setDaemon(true);

         myReceiver.setName("MongoDB " + mySocket.getLocalPort() + "<--"

                 + myServer.getCanonicalName());

     }

     /**

      * {@inheritDoc}

      */

     @Override

     public void close() throws IOException {

         if (myOpen.compareAndSet(true, false)) {

             myReceiver.interrupt();

             // Now that output is shutdown. Close up the socket. This

             // Triggers the receiver to close if the interrupt didn't work.

             myOutput.close();

             myInput.close();

             mySocket.close();

             try {

                 if (Thread.currentThread() != myReceiver) {

                     myReceiver.join();

                 }

             }

             catch (final InterruptedException ie) {

                 // Ignore.

             }

             myEventSupport.firePropertyChange(OPEN_PROP_NAME, true, false);

         }

     }

     /**

      * {@inheritDoc}

      * <p>

      * True if the send and pending queues are empty.

      * </p>

      */

     @Override

     public int getPendingCount() {

         return super.getPendingCount() + mySendSequence.getWaitersCount();

     }

     /**

      * {@inheritDoc}

      * <p>

      * True if the send and pending queues are empty.

      * </p>

      */

     @Override

     public boolean isIdle() {

         return super.isIdle() && mySendSequence.isIdle();

     }

     /**

      * {@inheritDoc}

      */

     @Override

     public void send(final Message message1, final Message message2,

             final ReplyCallback replyCallback) throws MongoDbException {

         validate(message1, message2);

         if (replyCallback instanceof AddressAware) {

             ((AddressAware) replyCallback).setAddress(myServer

                     .getCanonicalName());

         }

         final int count = (message2 == null) ? 1 : 2;

         final long seq = mySendSequence.reserve(count);

         final long end = seq + count;

         boolean sawError = false;

         final PendingMessage pendingMessage = new PendingMessage();

         try {

             // Serialize the messages now so the critical section becomes close

             // to a write(byte[]) (with a little accounting overhead).

             final Reference<BufferingBsonOutputStream> outRef = myBuffers.get();

             BufferingBsonOutputStream out = (outRef != null) ? outRef.get()

                     : null;

             if (out == null) {

                 out = new BufferingBsonOutputStream(

                         new RandomAccessOutputStream(myEncoderCache));

                 myBuffers

                         .set(new SoftReference<BufferingBsonOutputStream>(out));

             }

             message1.write((int) (seq & 0xFFFFFF), out);

             if (message2 != null) {

                 message2.write((int) ((seq + 1) & 0xFFFFFF), out);

             }

             // Now stand in line.

             mySendSequence.waitFor(seq);

             if (count == 1) {

                 pendingMessage.set((int) (seq & 0xFFFFFF), message1,

                         replyCallback);

                 send(pendingMessage, out.getOutput());

             }

             else {

                 pendingMessage.set((int) ((seq + 1) & 0xFFFFFF), message2,

                         replyCallback);

                 send(pendingMessage, out.getOutput());

             }

             // If no-one is waiting we need to flush the message.

             if (mySendSequence.noWaiter(end)) {

                 if (myReceiver != Thread.currentThread()) {

                     flush();

                 }

                 else {

                     markReaderNeedsToFlush();

                 }

             }

         }

         catch (final InterruptedException ie) {

             // Handled by loop but if we have a message, need to

             // tell him something bad happened (but we shouldn't).

             raiseError(ie, pendingMessage.getReplyCallback());

         }

         catch (final IOException ioe) {

             myLog.warn(ioe, "I/O Error sending a message.");

             raiseError(ioe, pendingMessage.getReplyCallback());

             sawError = true;

         }

         catch (final RuntimeException re) {

             myLog.warn(re, "Runtime error sending a message.");

             raiseError(re, pendingMessage.getReplyCallback());

             sawError = true;

         }

         catch (final Error error) {

             myLog.error(error, "Error sending a message.");

             raiseError(error, pendingMessage.getReplyCallback());

             sawError = true;

         }

         finally {

             pendingMessage.clear();

             mySendSequence.release(seq, end);

             if (sawError) {

                 // This may/will fail because we are dying.

                 try {

                     if (myOpen.get()) {

                         flush();

                     }

                 }

                 catch (final IOException ioe) {

                     myLog.warn(ioe, "I/O Error on final flush of messages.");

                 }

                 finally {

                     // Make sure we get shutdown completely.

                     IOUtils.close(SocketConnection.this);

                 }

             }

         }

     }

     /**

      * {@inheritDoc}

      */

     @Override

     public void send(final Message message, final ReplyCallback replyCallback)

             throws MongoDbException {

         send(message, null, replyCallback);

     }

     /**

      * Starts the connections read and write threads.

      */

     @Override

     public void start() {

         myReceiver.start();

         if (myServer.needBuildInfo()) {

             send(new BuildInfo(), new ServerUpdateCallback(myServer));

         }

     }

 }