/*
 * Copyright 2012 Metamarkets Group 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.metamx.metrics;

import com.metamx.common.concurrent.ScheduledExecutors;
import com.metamx.common.lifecycle.LifecycleStart;
import com.metamx.common.lifecycle.LifecycleStop;
import com.metamx.emitter.service.ServiceEmitter;

import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ScheduledExecutorService;

/**
 */
public class MonitorScheduler
{
  private final MonitorSchedulerConfig config;
  private final ScheduledExecutorService exec;
  private final ServiceEmitter emitter;
  private final List<Monitor> monitors;

  private volatile boolean started = false;

  public MonitorScheduler(
      MonitorSchedulerConfig config,
      ScheduledExecutorService exec,
      ServiceEmitter emitter,
      List<Monitor> monitors
  )
  {
    this.config = config;
    this.exec = exec;
    this.emitter = emitter;
    this.monitors = monitors;
  }

  @LifecycleStart
  public synchronized void start()
  {
    if (started) {
      return;
    }
    started = true;

    for (final Monitor monitor : monitors) {
      monitor.start();
      ScheduledExecutors.scheduleAtFixedRate(
          exec,
          config.getEmitterPeriod(),
          new Callable<ScheduledExecutors.Signal>()
          {
            @Override
            public ScheduledExecutors.Signal call() throws Exception
            {
              return monitor.monitor(emitter) ? ScheduledExecutors.Signal.REPEAT : ScheduledExecutors.Signal.STOP;
            }
          }
      );
    }
  }

  @LifecycleStop
  public synchronized void stop()
  {
    if (!started) {
      return;
    }

    started = false;
    for (Monitor monitor : monitors) {
      monitor.stop();
    }
  }
}

@@ -629,16 +629,86 @@
 trics;%0A%0A
+import com.google.common.collect.Lists;%0Aimport com.metamx.common.ISE;%0A
 import c
@@ -1547,24 +1547,43 @@
 itors = 
+Lists.newArrayList(
 monitors
 ;%0A  %7D%0A%0A 
@@ -1574,16 +1574,17 @@
 monitors
+)
 ;%0A  %7D%0A%0A 
@@ -1742,32 +1742,771 @@
 nitors) %7B%0A      
+startMonitor(monitor);%0A    %7D%0A  %7D%0A%0A  public synchronized void addMonitor(final Monitor monitor)%0A  %7B%0A    if (!started) %7B%0A      throw new ISE(%22addMonitor must be called after start%22);%0A    %7D%0A    if (hasMonitor(monitor)) %7B%0A      throw new ISE(%22Monitor already monitoring: %25s%22, monitor);%0A    %7D%0A    monitors.add(monitor);%0A    startMonitor(monitor);%0A  %7D%0A%0A  public synchronized void removeMonitor(final Monitor monitor)%0A  %7B%0A    monitors.remove(monitor);%0A    monitor.stop();%0A  %7D%0A%0A  @LifecycleStop%0A  public synchronized void stop()%0A  %7B%0A    if (!started) %7B%0A      return;%0A    %7D%0A%0A    started = false;%0A    for (Monitor monitor : monitors) %7B%0A      removeMonitor(monitor);%0A    %7D%0A  %7D%0A%0A  private synchronized void startMonitor(final Monitor monitor)%0A  %7B%0A    
 monitor.start();
@@ -2506,18 +2506,16 @@
 tart();%0A
-  
     Sche
@@ -2550,18 +2550,16 @@
 edRate(%0A
-  
         
@@ -2572,18 +2572,16 @@
         
-  
 config.g
@@ -2599,18 +2599,16 @@
 riod(),%0A
-  
         
@@ -2649,38 +2649,34 @@
 gnal%3E()%0A        
-  %7B%0A  
+%7B%0A
           @Overr
@@ -2679,18 +2679,16 @@
 verride%0A
-  
         
@@ -2752,26 +2752,24 @@
 n%0A          
-  
 %7B%0A          
@@ -2774,78 +2774,266 @@
     
-  return monitor.monitor(emitter) ? ScheduledExecutors.Signal.REPEAT :
+// Run one more time even if the monitor was removed, in case there's some extra data to flush%0A            if (monitor.monitor(emitter) && hasMonitor(monitor)) %7B%0A              return ScheduledExecutors.Signal.REPEAT;%0A            %7D else %7B%0A              return
  Sch
@@ -3093,32 +3093,34 @@
 %7D%0A      
-);
+  %7D
 %0A    
-%7D
+);
 %0A  %7D%0A%0A  
 @Lifecyc
@@ -3115,157 +3115,69 @@
 %0A%0A  
-@LifecycleStop%0A  public synchronized void stop()%0A  %7B%0A    if (!started) %7B%0A      return;%0A    %7D%0A%0A    started = false;%0A    for (Monitor m
+private synchronized boolean hasMonitor(final M
 onitor
- :
  monitor
 s) %7B
@@ -3176,26 +3176,32 @@
 itor
-s
 )
+%0A 
  %7B%0A    
- 
+return
  monitor
 .sto
@@ -3200,22 +3200,28 @@
 itor
-.stop();%0A    %7D
+s.contains(monitor);
 %0A  %7D

Allow runtime addition or removal of monitors