/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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.
 */
package com.datatorrent.bufferserver.server;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.Map.Entry;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.datatorrent.bufferserver.internal.DataList;
import com.datatorrent.bufferserver.internal.FastDataList;
import com.datatorrent.bufferserver.internal.LogicalNode;
import com.datatorrent.bufferserver.packet.PayloadTuple;
import com.datatorrent.bufferserver.packet.PublishRequestTuple;
import com.datatorrent.bufferserver.packet.PurgeRequestTuple;
import com.datatorrent.bufferserver.packet.ResetRequestTuple;
import com.datatorrent.bufferserver.packet.SubscribeRequestTuple;
import com.datatorrent.bufferserver.packet.Tuple;
import com.datatorrent.bufferserver.storage.Storage;
import com.datatorrent.common.util.NameableThreadFactory;
import com.datatorrent.netlet.AbstractLengthPrependerClient;
import com.datatorrent.netlet.AbstractServer;
import com.datatorrent.netlet.DefaultEventLoop;
import com.datatorrent.netlet.EventLoop;
import com.datatorrent.netlet.WriteOnlyLengthPrependerClient;
import com.datatorrent.netlet.util.VarInt;

/**
 * The buffer server application<p>
 * <br>
 *
 * @since 0.3.2
 */
public class Server extends AbstractServer
{
  public static final int DEFAULT_BUFFER_SIZE = 64 * 1024 * 1024;
  public static final int DEFAULT_NUMBER_OF_CACHED_BLOCKS = 8;
  private final int port;
  private String identity;
  private Storage storage;
  private final EventLoop eventloop;
  private final ExecutorService serverHelperExecutor;
  private final ExecutorService storageHelperExecutor;
  private volatile CountDownLatch latch;

  private byte[] authToken;

  private static final boolean BACK_PRESSURE_ENABLED = !Boolean.getBoolean("org.apache.apex.bufferserver.backpressure.disable");

  /**
   * @param port - port number to bind to or 0 to auto select a free port
   */
  public Server(EventLoop eventloop, int port)
  {
    this(eventloop, port, DEFAULT_BUFFER_SIZE, DEFAULT_NUMBER_OF_CACHED_BLOCKS);
  }

  public Server(EventLoop eventloop, int port, int blocksize, int numberOfCacheBlocks)
  {
    this.eventloop = eventloop;
    this.port = port;
    this.blockSize = blocksize;
    this.numberOfCacheBlocks = numberOfCacheBlocks;
    serverHelperExecutor = Executors.newSingleThreadExecutor(new NameableThreadFactory("ServerHelper"));
    final ArrayBlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(numberOfCacheBlocks);
    final NameableThreadFactory threadFactory = new NameableThreadFactory("StorageHelper");
    storageHelperExecutor = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, workQueue, threadFactory,
        new ThreadPoolExecutor.CallerRunsPolicy());
  }

  public void setSpoolStorage(Storage storage)
  {
    this.storage = storage;
  }

  @Override
  public void registered(SelectionKey key)
  {
    super.registered(key);
    logger.info("Server started listening at {}", getServerAddress());
    latch.countDown();
    latch = null;
  }

  @Override
  public void unregistered(SelectionKey key)
  {
    for (LogicalNode ln : subscriberGroups.values()) {
      ln.boot();
    }
    /*
     * There may be un-register tasks scheduled to run on the event loop that use serverHelperExecutor.
     */
    eventloop.submit(new Runnable()
    {
      @Override
      public void run()
      {
        serverHelperExecutor.shutdown();
        storageHelperExecutor.shutdown();
        try {
          serverHelperExecutor.awaitTermination(5000, TimeUnit.MILLISECONDS);
          storageHelperExecutor.awaitTermination(5000, TimeUnit.MILLISECONDS);
        } catch (InterruptedException ex) {
          logger.debug("Executor Termination", ex);
        }
        logger.info("Server stopped listening at {}", getServerAddress());
        latch.countDown();
        latch = null;
      }
    });
  }

  public InetSocketAddress run()
  {
    final CountDownLatch latch = new CountDownLatch(1);
    this.latch = latch;
    eventloop.start(null, port, this);
    try {
      latch.await();
    } catch (InterruptedException e) {
      throw new RuntimeException(e);
    }
    return (InetSocketAddress)getServerAddress();
  }

  public InetSocketAddress run(long time)
  {
    if (time < 0) {
      throw new IllegalArgumentException(String.format("Wait time %d can't be negative", time));
    }
    final CountDownLatch latch = new CountDownLatch(1);
    this.latch = latch;
    eventloop.start(null, port, this);
    final long deadline = System.currentTimeMillis() + time;
    try {
      while (latch.getCount() != 0 && time > 0 && latch.await(time, TimeUnit.MILLISECONDS)) {
        time = deadline - System.currentTimeMillis();
      }
    } catch (InterruptedException e) {
      throw new RuntimeException(e);
    }
    return (InetSocketAddress)getServerAddress();
  }

  public void stop()
  {
    final CountDownLatch latch = new CountDownLatch(1);
    this.latch = latch;
    eventloop.stop(this);
    try {
      latch.await();
    } catch (InterruptedException e) {
      throw new RuntimeException(e);
    } finally {
      shutdownExecutors(latch.getCount() == 0);
    }
  }

  public void stop(long time)
  {
    if (time < 0) {
      throw new IllegalArgumentException(String.format("Wait time %d can't be negative", time));
    }
    final CountDownLatch latch = new CountDownLatch(1);
    this.latch = latch;
    eventloop.stop(this);
    final long deadline = System.currentTimeMillis() + time;
    try {
      while (latch.getCount() != 0 && time > 0 && latch.await(time, TimeUnit.MILLISECONDS)) {
        time = deadline - System.currentTimeMillis();
      }
    } catch (InterruptedException e) {
      throw new RuntimeException(e);
    } finally {
      shutdownExecutors(latch.getCount() == 0);
    }
  }

  private void shutdownExecutors(boolean isTerminated)
  {
    if (!isTerminated) {
      logger.warn("Buffer server {} did not terminate.", this);
      try {
        if (!serverHelperExecutor.isTerminated()) {
          logger.warn("Forcing termination of {}", serverHelperExecutor);
          serverHelperExecutor.shutdownNow();
        }
        if (!storageHelperExecutor.isTerminated()) {
          logger.warn("Forcing termination of {}", storageHelperExecutor);
          storageHelperExecutor.shutdownNow();
        }
      } catch (RuntimeException e) {
        logger.error("Exception while terminating executors", e);
      }
    }
  }

  public void setAuthToken(byte[] authToken)
  {
    this.authToken = authToken;
  }

  /**
   *
   * @param args
   * @throws Exception
   */
  public static void main(String[] args) throws Exception
  {
    int port;
    if (args.length > 0) {
      port = Integer.parseInt(args[0]);
    } else {
      port = 0;
    }

    DefaultEventLoop eventloop = DefaultEventLoop.createEventLoop("alone");
    Thread thread = eventloop.start();
    new Server(eventloop, port).run();
    thread.join();
  }

  @Override
  public String toString()
  {
    return getClass().getSimpleName() + '@' + Integer.toHexString(hashCode()) + "{address=" + getServerAddress() + "}";
  }

  private final ConcurrentHashMap<String, DataList> publisherBuffers = new ConcurrentHashMap<>(1, 0.75f, 1);
  private final ConcurrentHashMap<String, LogicalNode> subscriberGroups = new ConcurrentHashMap<>();
  private final ConcurrentHashMap<String, AbstractLengthPrependerClient> publisherChannels = new ConcurrentHashMap<>();
  private final int blockSize;
  private final int numberOfCacheBlocks;

  private void handlePurgeRequest(PurgeRequestTuple request, final AbstractLengthPrependerClient ctx) throws IOException
  {
    DataList dl;
    dl = publisherBuffers.get(request.getIdentifier());

    byte[] message;
    if (dl == null) {
      message = ("Invalid identifier '" + request.getIdentifier() + "'").getBytes();
    } else {
      dl.purge((long)request.getBaseSeconds() << 32 | request.getWindowId());
      message = ("Request sent for processing: " + request).getBytes();
    }

    final byte[] tuple = PayloadTuple.getSerializedTuple(0, message.length);
    System.arraycopy(message, 0, tuple, tuple.length - message.length, message.length);
    if (ctx.write(tuple)) {
      ctx.write();
    } else {
      logger.error("Failed to deliver purge ack message. {} send buffers are full.", ctx);
      throw new RuntimeException("Failed to deliver purge ack message. " + ctx + "send buffers are full.");
    }
  }

  public void purge(long windowId)
  {
    for (DataList dataList: publisherBuffers.values()) {
      dataList.purge(windowId);
    }
  }

  private void handleResetRequest(ResetRequestTuple request, final AbstractLengthPrependerClient ctx) throws IOException
  {
    DataList dl;
    dl = publisherBuffers.remove(request.getIdentifier());

    byte[] message;
    if (dl == null) {
      message = ("Invalid identifier '" + request.getIdentifier() + "'").getBytes();
    } else {
      AbstractLengthPrependerClient channel = publisherChannels.remove(request.getIdentifier());
      if (channel != null) {
        eventloop.disconnect(channel);
      }
      dl.reset();
      message = ("Request sent for processing: " + request).getBytes();
    }

    final byte[] tuple = PayloadTuple.getSerializedTuple(0, message.length);
    System.arraycopy(message, 0, tuple, tuple.length - message.length, message.length);
    if (ctx.write(tuple)) {
      ctx.write();
    } else {
      logger.error("Failed to deliver reset ack message. {} send buffers are full.", ctx);
      throw new RuntimeException("Failed to deliver reset ack message. " + ctx + "send buffers are full.");
    }
  }

  /**
   *
   * @param request
   * @param key
   */
  private void handleSubscriberRequest(final SubscribeRequestTuple request, final SelectionKey key)
  {
    try {
      serverHelperExecutor.submit(new Runnable()
      {
        @Override
        public void run()
        {
          final String upstream_identifier = request.getUpstreamIdentifier();

          /*
           * if there is already a datalist registered for the type in which this client is interested,
           * then get a iterator on the data items of that data list. If the datalist is not registered,
           * then create one and register it. Hopefully this one would be used by future upstream nodes.
           */
          DataList dl = publisherBuffers.get(upstream_identifier);
          if (dl == null) {
            dl = Tuple.FAST_VERSION.equals(request.getVersion()) ?
                new FastDataList(upstream_identifier, blockSize, numberOfCacheBlocks, BACK_PRESSURE_ENABLED) :
                new DataList(upstream_identifier, blockSize, numberOfCacheBlocks, BACK_PRESSURE_ENABLED);
            DataList odl = publisherBuffers.putIfAbsent(upstream_identifier, dl);
            if (odl != null) {
              dl = odl;
            }
          }

          final String identifier = request.getIdentifier();
          final String type = request.getStreamType();
          final long skipWindowId = (long)request.getBaseSeconds() << 32 | request.getWindowId();
          final LogicalNode ln = new LogicalNode(identifier, upstream_identifier, type, dl
              .newIterator(skipWindowId), skipWindowId, eventloop);

          int mask = request.getMask();
          if (mask != 0) {
            for (Integer bs : request.getPartitions()) {
              ln.addPartition(bs, mask);
            }
          }
          final LogicalNode oln = subscriberGroups.put(type, ln);
          if (oln != null) {
            oln.boot();
          }
          final Subscriber subscriber = new Subscriber(ln, request.getBufferSize());
          eventloop.submit(new Runnable()
          {
            @Override
            public void run()
            {
              key.attach(subscriber);
              subscriber.registered(key);
              subscriber.connected();
            }
          });
        }
      });
    } catch (RejectedExecutionException e) {
      logger.error("Received subscriber request {} after server {} termination. Disconnecting {}.", request, this, key.channel(), e);
      if (key.isValid()) {
        try {
          key.channel().close();
        } catch (IOException ioe) {
          logger.error("Failed to close channel {}", key.channel(), ioe);
        }
      }
    }
  }

  private void handleSubscriberTeardown(final SelectionKey key)
  {
    try {
      final Subscriber subscriber = (Subscriber)key.attachment();
      if (subscriber != null) {
        serverHelperExecutor.submit(new Runnable()
        {
          @Override
          public void run()
          {
            try {
              final LogicalNode ln = subscriber.ln;
              if (ln != null) {
                ln.removeChannel(subscriber);
                if (ln.getPhysicalNodeCount() == 0) {
                  DataList dl = publisherBuffers.get(ln.getUpstream());
                  if (dl != null) {
                    logger.info("Removing ln {} from dl {}", ln, dl);
                    dl.removeDataListener(ln);
                  }
                  subscriberGroups.remove(ln.getGroup(), ln);
                  ln.getIterator().close();
                }
                subscriber.ln = null;
              }
            } catch (Throwable t) {
              logger.error("Buffer server {} failed to tear down subscriber {}.", Server.this, subscriber, t);
            }
          }

          @Override
          public String toString()
          {
            return subscriber + " teardown task.";
          }
        });
      } else {
        logger.error("Selection key {} has unexpected attachment {}.", key, key.attachment());
      }
    } catch (ClassCastException e) {
      logger.error("Selection key {} has unexpected attachment {}.", key, key.attachment());
    } catch (RejectedExecutionException e) {
      logger.error("Subscriber {} teardown after server {} termination.", key.attachment(), this, e);
    }
  }

  /**
   *
   * @param request
   * @param connection
   * @return
   */
  public DataList handlePublisherRequest(PublishRequestTuple request, AbstractLengthPrependerClient connection)
  {
    String identifier = request.getIdentifier();

    DataList dl = publisherBuffers.get(identifier);

    if (dl != null) {
      /*
       * close previous connection with the same identifier which is guaranteed to be unique.
       */
      AbstractLengthPrependerClient previous = publisherChannels.put(identifier, connection);
      if (previous != null) {
        eventloop.disconnect(previous);
      }

      try {
        dl.rewind(request.getBaseSeconds(), request.getWindowId());
      } catch (IOException ie) {
        throw new RuntimeException(ie);
      }
    } else {
      dl = Tuple.FAST_VERSION.equals(request.getVersion()) ?
          new FastDataList(identifier, blockSize, numberOfCacheBlocks, BACK_PRESSURE_ENABLED) :
          new DataList(identifier, blockSize, numberOfCacheBlocks, BACK_PRESSURE_ENABLED);
      DataList odl = publisherBuffers.putIfAbsent(identifier, dl);
      if (odl != null) {
        dl = odl;
      }
    }
    dl.setSecondaryStorage(storage, storageHelperExecutor);

    return dl;
  }

  @Override
  public ClientListener getClientConnection(SocketChannel sc, ServerSocketChannel ssc)
  {
    ClientListener client;
    if (authToken == null) {
      client = new UnidentifiedClient();
    } else {
      AuthClient authClient = new AuthClient();
      authClient.setToken(authToken);
      client = authClient;
    }
    return client;
  }

  @Override
  public void handleException(Exception cce, EventLoop el)
  {
    if (cce instanceof RuntimeException) {
      throw (RuntimeException)cce;
    }

    throw new RuntimeException(cce);
  }

  class AuthClient extends com.datatorrent.bufferserver.client.AuthClient
  {
    boolean ignore;

    @Override
    public void onMessage(byte[] buffer, int offset, int size)
    {
      if (ignore) {
        return;
      }

      authenticateMessage(buffer, offset, size);

      unregistered(key);
      UnidentifiedClient client = new UnidentifiedClient();
      key.attach(client);
      key.interestOps(SelectionKey.OP_READ);
      client.registered(key);
      client.connected();

      int len = writeOffset - readOffset - size;
      if (len > 0) {
        client.transferBuffer(buffer, readOffset + size, len);
      }

      ignore = true;
    }
  }

  class UnidentifiedClient extends SeedDataClient
  {
    boolean ignore;

    @Override
    public void onMessage(byte[] buffer, int offset, int size)
    {
      if (ignore) {
        return;
      }

      Tuple request = Tuple.getTuple(buffer, offset, size);
      switch (request.getType()) {
        case PUBLISHER_REQUEST:

          /*
           * unregister the unidentified client since its job is done!
           */
          unregistered(key);
          logger.info("Received publisher request: {}", request);
          PublishRequestTuple publisherRequest = (PublishRequestTuple)request;

          DataList dl = handlePublisherRequest(publisherRequest, this);
          dl.setAutoFlushExecutor(serverHelperExecutor);

          Publisher publisher;
          if (publisherRequest.getVersion().equals(Tuple.FAST_VERSION)) {
            publisher = new Publisher(dl, (long)request.getBaseSeconds() << 32 | request.getWindowId())
            {
              @Override
              public int readSize()
              {
                if (writeOffset - readOffset < 2) {
                  return -1;
                }

                short s = buffer[readOffset++];
                return s | (buffer[readOffset++] << 8);
              }

            };
          } else {
            publisher = new Publisher(dl, (long)request.getBaseSeconds() << 32 | request.getWindowId());
          }

          key.attach(publisher);
          key.interestOps(SelectionKey.OP_READ);
          publisher.registered(key);

          int len = writeOffset - readOffset - size;
          if (len > 0) {
            publisher.transferBuffer(this.buffer, readOffset + size, len);
          }
          ignore = true;

          break;

        case SUBSCRIBER_REQUEST:
          /*
           * unregister the unidentified client since its job is done!
           */
          unregistered(key.interestOps(0));
          ignore = true;
          logger.info("Received subscriber request: {}", request);

          handleSubscriberRequest((SubscribeRequestTuple)request, key);
          break;

        case PURGE_REQUEST:
          logger.info("Received purge request: {}", request);
          try {
            handlePurgeRequest((PurgeRequestTuple)request, this);
          } catch (IOException io) {
            throw new RuntimeException(io);
          }
          break;

        case RESET_REQUEST:
          logger.info("Received reset all request: {}", request);
          try {
            handleResetRequest((ResetRequestTuple)request, this);
          } catch (IOException io) {
            throw new RuntimeException(io);
          }
          break;

        default:
          throw new RuntimeException("unexpected message: " + request.toString());
      }
    }

  }

  private class Subscriber extends WriteOnlyLengthPrependerClient
  {
    private LogicalNode ln;

    Subscriber(LogicalNode ln, int bufferSize)
    {
      super(1024 * 1024, bufferSize == 0 ? 256 * 1024 : bufferSize);
      this.ln = ln;
      ln.addConnection(this);
    }

    @Override
    public void connected()
    {
      super.connected();
      serverHelperExecutor.submit(new Runnable()
      {
        @Override
        public void run()
        {
          final DataList dl = publisherBuffers.get(ln.getUpstream());
          if (dl != null) {
            ln.catchUp();
            dl.addDataListener(ln);
          } else {
            logger.error("Disconnecting {} with no matching data list.", this);
            ln.boot();
          }
        }
      });
    }

    @Override
    public void unregistered(final SelectionKey key)
    {
      handleSubscriberTeardown(key);
      super.unregistered(key);
    }

    @Override
    public String toString()
    {
      return getClass().getSimpleName() + '@' + Integer.toHexString(hashCode()) + "{ln=" + ln + "}";
    }
  }

  /**
   * When the publisher connects to the server and starts publishing the data,
   * this is the end on the server side which handles all the communication.
   *
   */
  class Publisher extends SeedDataClient
  {
    private final DataList datalist;
    boolean dirty;

    Publisher(DataList dl, long windowId)
    {
      super(dl.getBuffer(windowId), dl.getPosition(), 1024);
      this.datalist = dl;
    }

    @Override
    public void onMessage(byte[] buffer, int offset, int size)
    {
      //if (buffer[offset] == MessageType.BEGIN_WINDOW_VALUE || buffer[offset] == MessageType.END_WINDOW_VALUE) {
      //  logger.debug("server received {}", Tuple.getTuple(buffer, offset, size));
      //}
      dirty = true;
    }

    /**
     * Schedules a task to conditionally resume I/O channel read operations.
     * No-op if {@linkplain java.nio.channels.SelectionKey#OP_READ OP_READ}
     * is already set in the key {@linkplain java.nio.channels.SelectionKey#interestOps() interestOps}.
     * Otherwise, calls {@linkplain #read(int) read(0)} to process data
     * left in the Publisher read buffer and registers {@linkplain java.nio.channels.SelectionKey#OP_READ OP_READ}
     * in the key {@linkplain java.nio.channels.SelectionKey#interestOps() interestOps}.
     * @return true
     */
    @Override
    public boolean resumeReadIfSuspended()
    {
      eventloop.submit(new Runnable()
      {
        @Override
        public void run()
        {
          final int interestOps = key.interestOps();
          if ((interestOps & SelectionKey.OP_READ) == 0) {
            if (readExt(0)) {
              logger.debug("Resuming read on key {} with attachment {}", key, key.attachment());
              key.interestOps(interestOps | SelectionKey.OP_READ);
            } else {
              logger.debug("Keeping read on key {} with attachment {} suspended. ", key, key.attachment(), datalist);
              datalist.notifyListeners();
            }
          }
        }
      });
      return true;
    }

    @Override
    public void read(int len)
    {
      readExt(len);
    }

    private boolean readExt(int len)
    {
      //logger.debug("read {} bytes", len);
      writeOffset += len;
      do {
        if (size <= 0) {
          switch (size = readSize()) {
            case -1:
              if (writeOffset == buffer.length) {
                if (readOffset > writeOffset - 5) {
                  dirty = false;
                  datalist.flush(writeOffset);
                  /*
                   * if the data is not corrupt, we are limited by space to receive full varint.
                   * so we allocate a new byteBuffer and copy over the partially written data to the
                   * new byteBuffer and start as if we always had full room but not enough data.
                   */
                  if (!switchToNewBufferOrSuspendRead(buffer, readOffset, size  + VarInt.getSize(size))) {
                    return false;
                  }
                }
              } else if (dirty) {
                dirty = false;
                datalist.flush(writeOffset);
              }
              return true;

            case 0:
              continue;

            default:
              break;
          }
        }

        if (writeOffset - readOffset >= size) {
          onMessage(buffer, readOffset, size);
          readOffset += size;
          size = 0;
        } else {
          if (writeOffset == buffer.length) {
            dirty = false;
            datalist.flush(writeOffset);
            /*
             * hit wall while writing serialized data, so have to allocate a new byteBuffer.
             */
            if (!switchToNewBufferOrSuspendRead(buffer, readOffset - VarInt.getSize(size), size + VarInt.getSize(size))) {
              readOffset -= VarInt.getSize(size);
              size = 0;
              return false;
            }
            size = 0;
          } else if (dirty) {
            dirty = false;
            datalist.flush(writeOffset);
          }
          return true;
        }
      } while (true);
    }

    private boolean switchToNewBufferOrSuspendRead(final byte[] array, final int offset, final int size)
    {
      if (switchToNewBuffer(array, offset, size)) {
        return true;
      }
      datalist.suspendRead(this);
      return false;
    }

    private boolean switchToNewBuffer(final byte[] array, final int offset, final int size)
    {
      if (datalist.isMemoryBlockAvailable()) {
        final byte[] newBuffer = datalist.newBuffer(size);
        byteBuffer = ByteBuffer.wrap(newBuffer);
        if (array == null || array.length - offset == 0) {
          writeOffset = 0;
        } else {
          writeOffset = array.length - offset;
          System.arraycopy(buffer, offset, newBuffer, 0, writeOffset);
          byteBuffer.position(writeOffset);
        }
        buffer = newBuffer;
        readOffset = 0;
        datalist.addBuffer(buffer);
        return true;
      }
      return false;
    }

    @Override
    public void unregistered(final SelectionKey key)
    {
      super.unregistered(key);
      teardown();
    }

    @Override
    public void handleException(Exception cce, EventLoop el)
    {
      teardown();

      if (cce instanceof RejectedExecutionException && serverHelperExecutor.isTerminated()) {
        logger.warn("Terminated Executor Exception for {}.", this, cce);
        el.disconnect(this);
      } else {
        super.handleException(cce, el);
      }
    }

    @Override
    public String toString()
    {
      return getClass().getName() + '@' + Integer.toHexString(hashCode()) + " {datalist=" + datalist + '}';
    }

    private volatile boolean torndown;

    private void teardown()
    {
      //logger.debug("Teardown is being called {}", torndown, new Exception());
      if (torndown) {
        return;
      }
      torndown = true;

      /*
       * if the publisher unregistered, all the downstream guys are going to be unregistered anyways
       * in our world. So it makes sense to kick them out proactively. Otherwise these clients since
       * are not being written to, just stick around till the next publisher shows up and eat into
       * the data it's publishing for the new subscribers.
       */

      /**
       * since the publisher server died, the queue which it was using would stop pumping the data unless
       * a new publisher comes up with the same name. We leave it to the stream to decide when to bring up a new node
       * with the same identifier as the one which just died.
       */
      if (publisherChannels.containsValue(this)) {
        final Iterator<Entry<String, AbstractLengthPrependerClient>> i = publisherChannels.entrySet().iterator();
        while (i.hasNext()) {
          if (i.next().getValue() == this) {
            i.remove();
            break;
          }
        }
      }

      ArrayList<LogicalNode> list = new ArrayList<>();
      String publisherIdentifier = datalist.getIdentifier();
      Iterator<LogicalNode> iterator = subscriberGroups.values().iterator();
      while (iterator.hasNext()) {
        LogicalNode ln = iterator.next();
        if (publisherIdentifier.equals(ln.getUpstream())) {
          list.add(ln);
        }
      }

      for (LogicalNode ln : list) {
        ln.boot();
      }
    }

  }

  abstract class SeedDataClient extends AbstractLengthPrependerClient
  {

    public SeedDataClient()
    {
    }

    public SeedDataClient(int readBufferSize, int sendBufferSize)
    {
      super(readBufferSize, sendBufferSize);
    }

    public SeedDataClient(byte[] readbuffer, int position, int sendBufferSize)
    {
      super(readbuffer, position, sendBufferSize);
    }

    public void transferBuffer(byte[] array, int offset, int len)
    {
      int remainingCapacity;
      do {
        remainingCapacity = buffer.length - writeOffset;
        if (len < remainingCapacity) {
          remainingCapacity = len;
          byteBuffer.position(writeOffset + remainingCapacity);
        } else {
          byteBuffer.position(buffer.length);
        }
        System.arraycopy(array, offset, buffer, writeOffset, remainingCapacity);
        read(remainingCapacity);

        offset += remainingCapacity;
      } while ((len -= remainingCapacity) > 0);
    }
  }

  private static final Logger logger = LoggerFactory.getLogger(Server.class);
}