/**
 * Copyright 2012 Netflix, 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.
 */
package com.netflix.servo.examples;

import com.google.common.io.ByteStreams;
import com.google.common.io.Closeables;
import com.google.common.io.CountingInputStream;
import com.google.common.io.CountingOutputStream;

import com.netflix.servo.publish.BasicMetricFilter;
import com.netflix.servo.publish.CounterToRateMetricTransform;
import com.netflix.servo.publish.FileMetricObserver;
import com.netflix.servo.publish.MetricObserver;
import com.netflix.servo.publish.MonitorRegistryMetricPoller;
import com.netflix.servo.publish.PollRunnable;
import com.netflix.servo.publish.PollScheduler;

import com.netflix.servo.tag.SortedTagList;
import com.netflix.servo.tag.TagList;
import com.netflix.servo.tag.Tags;

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

import java.io.File;
import java.io.IOException;

import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketAddress;

import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * An really basic echo server that uses the utility methods from
 * {@link com.netflix.servo.util.Counters}.
 */
public class EchoServerExample {

    private static final Logger LOGGER = LoggerFactory.getLogger(EchoServerExample.class);
    public static final int DEFAULT_PORT = 54321;

    private final int port;

    public EchoServerExample(int port) {
        this.port = port;
    }

    public void start() throws IOException {
        AcceptTask task = new AcceptTask(port);
        Thread t = new Thread(task, "AcceptTask");
        t.start();
    }

    public void shutdown() {
        // Just an example, a real server should have a way to cleanly
        // shutdown
    }

    public static class AcceptTask implements Runnable {
        private static final String[] COUNTRIES = {"US", "CA", "GB", "IE"};
        private final ServerSocket ss;
        private final Random r = new Random();

        public AcceptTask(int port) throws IOException {
            ss = new ServerSocket(port);
        }

        public TagList getTags(Socket s) {
            String country = COUNTRIES[r.nextInt(COUNTRIES.length)];
            return SortedTagList.builder().withTag(Tags.newTag("Country", country)).build();
        }

        public void run() {
            while (true) {
                try {
                    Socket s = ss.accept();
                    TagList tags = getTags(s);
                    LOGGER.info("received connection from {} with tags {}",
                        s.getRemoteSocketAddress(), tags);

                    //Counters.increment("RequestCount", tags);
                    ClientTask task = new ClientTask(tags, s);
                    Thread t = new Thread(task, "ClientTask");
                    t.start();
                } catch (IOException e) {
                    LOGGER.error("failure accepting connection", e);
                }
            }
        }
    }

    public static class ClientTask implements Runnable {
        private final TagList tags;
        private final Socket s;

        public ClientTask(TagList tags, Socket s) {
            this.tags = tags;
            this.s = s;
        }

        private void doWork() throws IOException {
            CountingInputStream input = null;
            CountingOutputStream output = null;
            try {
                input = new CountingInputStream(s.getInputStream());
                output = new CountingOutputStream(s.getOutputStream());
                ByteStreams.copy(input, output);
                //Counters.increment("BytesIn", input.getCount());
                //Counters.increment("BytesOut", output.getCount());
            } finally {
                Closeables.closeQuietly(input);
                Closeables.closeQuietly(output);
            }
        }

        public void run() {
            // Setup context so all counters increments in this thread will get
            // tagged
            //TaggingContext.setTags(tags);

            try {
                doWork();
            } catch (IOException e) {
                SocketAddress a = s.getRemoteSocketAddress();
                LOGGER.error("failure handling connection from " + a, e);
            }
        }
    }

    public static void main(String[] args) throws Exception {
        // Schedule collection of monitor registry metrics every 10 seconds
        PollScheduler scheduler = PollScheduler.getInstance();
        scheduler.start();

        final int heartbeatInterval = 20;
        MetricObserver transform = new CounterToRateMetricTransform(
            new FileMetricObserver("serverstat", new File(".")),
            heartbeatInterval, TimeUnit.SECONDS);

        PollRunnable task = new PollRunnable(
            new MonitorRegistryMetricPoller(),
            BasicMetricFilter.MATCH_ALL,
            transform);

        final int samplingInterval = 10;
        scheduler.addPoller(task, samplingInterval, TimeUnit.SECONDS);

        // Run server
        int port = DEFAULT_PORT;
        if (args.length > 0) {
            port = Integer.valueOf(args[0]);
        }
        EchoServerExample example = new EchoServerExample(port);
        example.start();
    }
}

@@ -808,17 +808,65 @@
 Stream;%0A
+import com.netflix.servo.monitor.DynamicCounter;
 %0A
-
 import c
@@ -1231,17 +1231,16 @@
 eduler;%0A
-%0A
 import c
@@ -1260,22 +1260,21 @@
 rvo.tag.
-Sorted
+Basic
 TagList;
@@ -1316,44 +1316,8 @@
 st;%0A
-import com.netflix.servo.tag.Tags;%0A%0A
 impo
@@ -1419,17 +1419,16 @@
 eption;%0A
-%0A
 import j
@@ -1507,34 +1507,8 @@
 ess;
-%0A%0Aimport java.util.Random;
 %0Aimp
@@ -1645,23 +1645,31 @@
 rvo.
-util.
+monitor.Dynamic
 Counter
-s%7D.
+%7D
 %0A */
@@ -2420,55 +2420,8 @@
  ss;
-%0A        private final Random r = new Random();
 %0A%0A  
@@ -2586,45 +2586,81 @@
     
-String country = COUNTRIES%5Br.nextInt(
+// Find the remote country by using this socket%0A            int length = 
 COUN
@@ -2675,10 +2675,8 @@
 ngth
-)%5D
 ;%0A  
@@ -2689,58 +2689,225 @@
     
-return SortedTagList.builder().withTag(Tags.newTag
+// force countryCode to be non-negative%0A            int countryCode = (s.getInetAddress().hashCode() %25 length + length) %25 length;%0A            String country = COUNTRIES%5BcountryCode%5D;%0A            return BasicTagList.of
 (%22Co
@@ -2926,17 +2926,8 @@
 try)
-).build()
 ;%0A  
@@ -3255,34 +3255,38 @@
             
-//
+Dynamic
 Counter
-s
 .increment(%22
@@ -4232,34 +4232,38 @@
             
-//
+Dynamic
 Counter
-s
 .increment(%22
@@ -4271,16 +4271,22 @@
 ytesIn%22,
+ tags,
  input.g
@@ -4317,18 +4317,22 @@
     
-//
+Dynamic
 Counter
-s
 .inc
@@ -4349,16 +4349,22 @@
 tesOut%22,
+ tags,
  output.

Make the example use the DynamicCounter utility class.