// This file is part of OpenTSDB.
// Copyright (C) 2010-2012  The OpenTSDB Authors.
//
// This program 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 program 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 program.  If not,
// see <http://www.gnu.org/licenses/>.
package net.opentsdb.tsd;

import java.io.IOException;
import java.nio.channels.ClosedChannelException;
import java.util.concurrent.atomic.AtomicLong;

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

import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelEvent;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.SimpleChannelHandler;
import org.jboss.netty.channel.group.DefaultChannelGroup;

import net.opentsdb.stats.StatsCollector;

/**
 * Keeps track of all existing connections.
 */
final class ConnectionManager extends SimpleChannelHandler {

  private static final Logger LOG = LoggerFactory.getLogger(ConnectionManager.class);

  private static final AtomicLong connections_established = new AtomicLong();
  private static final AtomicLong exceptions_caught = new AtomicLong();

  private static final DefaultChannelGroup channels =
    new DefaultChannelGroup("all-channels");

  static void closeAllConnections() {
    channels.close().awaitUninterruptibly();
  }

  /** Constructor. */
  public ConnectionManager() {
  }

  /**
   * Collects the stats and metrics tracked by this instance.
   * @param collector The collector to use.
   */
  public static void collectStats(final StatsCollector collector) {
    collector.record("connectionmgr.connections", channels.size(), "type=open");
    collector.record("connectionmgr.connections", connections_established, 
        "type=total");
    collector.record("connectionmgr.exceptions", exceptions_caught);
  }

  @Override
  public void channelOpen(final ChannelHandlerContext ctx,
                          final ChannelStateEvent e) {
    channels.add(e.getChannel());
    connections_established.incrementAndGet();
  }

  @Override
  public void handleUpstream(final ChannelHandlerContext ctx,
                             final ChannelEvent e) throws Exception {
    if (e instanceof ChannelStateEvent) {
      LOG.info(e.toString());
    }
    super.handleUpstream(ctx, e);
  }

  @Override
  public void exceptionCaught(final ChannelHandlerContext ctx,
                              final ExceptionEvent e) {
    final Throwable cause = e.getCause();
    final Channel chan = ctx.getChannel();
    exceptions_caught.incrementAndGet();
    if (cause instanceof ClosedChannelException) {
      LOG.warn("Attempt to write to closed channel " + chan);
      return;
    }
    if (cause instanceof IOException) {
      final String message = cause.getMessage();
      if ("Connection reset by peer".equals(message)
          || "Connection timed out".equals(message)) {
        // Do nothing.  A client disconnecting isn't really our problem.  Oh,
        // and I'm not kidding you, there's no better way to detect ECONNRESET
        // in Java.  Like, people have been bitching about errno for years,
        // and Java managed to do something *far* worse.  That's quite a feat.
        return;
      }
    }
    LOG.error("Unexpected exception from downstream for " + chan, cause);
    e.getChannel().close();
  }

}

@@ -1667,21 +1667,238 @@
 eptions_
-caugh
+unknown = new AtomicLong();%0A  private static final AtomicLong exceptions_closed = new AtomicLong();%0A  private static final AtomicLong exceptions_reset = new AtomicLong();%0A  private static final AtomicLong exceptions_timeou
 t = new 
@@ -2585,21 +2585,326 @@
 ptions_c
-aught
+losed, %0A        %22type=closed%22);%0A    collector.record(%22connectionmgr.exceptions%22, exceptions_reset, %0A        %22type=reset%22);%0A    collector.record(%22connectionmgr.exceptions%22, exceptions_timeout, %0A        %22type=timeout%22);%0A    collector.record(%22connectionmgr.exceptions%22, exceptions_unknown, %0A        %22type=unknown%22
 );%0A  %7D%0A%0A
@@ -3596,49 +3596,8 @@
 ();%0A
-    exceptions_caught.incrementAndGet();%0A
     
@@ -3639,24 +3639,67 @@
 xception) %7B%0A
+      exceptions_closed.incrementAndGet();%0A
       LOG.wa
@@ -3909,16 +3909,19 @@
 message)
+) %7B
 %0A       
@@ -3925,54 +3925,164 @@
     
-  %7C%7C %22Connection timed out%22.equals(message)) %7B
+exceptions_reset.incrementAndGet();%0A        return;%0A      %7D else if (%22Connection timed out%22.equals(message)) %7B%0A        exceptions_timeout.incrementAndGet();
 %0A   
@@ -4414,22 +4414,64 @@
       %7D%0A
-
     %7D%0A
+    exceptions_unknown.incrementAndGet();%0A
     LOG.

Split ConnectionManager exception types into unknown, closed, reset and timeout for greater detail into problems