code: /*
   * 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 org.apache.skywalking.oap.server.library.datacarrier.consumer;
  
  import java.lang.reflect.InvocationTargetException;
  import java.util.Properties;
  import java.util.concurrent.locks.ReentrantLock;
  import org.apache.skywalking.oap.server.library.datacarrier.buffer.Channels;
  
  /**
   * Pool of consumers <p> Created by wusheng on 2016/10/25.
   */
  public class ConsumeDriver<T> implements IDriver {
      private boolean running;
      private ConsumerThread[] consumerThreads;
      private Channels<T> channels;
      private ReentrantLock lock;
  
      public ConsumeDriver(String name,
                           Channels<T> channels, Class<? extends IConsumer<T>> consumerClass,
                           int num,
                           long consumeCycle,
                           Properties properties) {
          this(channels, num);
          for (int i = 0; i < num; i++) {
              consumerThreads[i] = new ConsumerThread(
                  "DataCarrier." + name + ".Consumer." + i + ".Thread", getNewConsumerInstance(consumerClass, properties),
                  consumeCycle
              );
              consumerThreads[i].setDaemon(true);
          }
      }
  
      public ConsumeDriver(String name, Channels<T> channels, IConsumer<T> prototype, int num, long consumeCycle) {
          this(channels, num);
          prototype.init(new Properties());
          for (int i = 0; i < num; i++) {
              consumerThreads[i] = new ConsumerThread(
                  "DataCarrier." + name + ".Consumer." + i + ".Thread", prototype, consumeCycle);
              consumerThreads[i].setDaemon(true);
          }
  
      }
  
      private ConsumeDriver(Channels<T> channels, int num) {
          running = false;
          this.channels = channels;
          consumerThreads = new ConsumerThread[num];
          lock = new ReentrantLock();
      }
  
      private IConsumer<T> getNewConsumerInstance(Class<? extends IConsumer<T>> consumerClass, Properties properties) {
          try {
              IConsumer<T> inst = consumerClass.getDeclaredConstructor().newInstance();
              inst.init(properties);
              return inst;
          } catch (InstantiationException e) {
              throw new ConsumerCannotBeCreatedException(e);
          } catch (IllegalAccessException e) {
              throw new ConsumerCannotBeCreatedException(e);
          } catch (NoSuchMethodException e) {
              throw new ConsumerCannotBeCreatedException(e);
          } catch (InvocationTargetException e) {
              throw new ConsumerCannotBeCreatedException(e);
          }
      }
  
      @Override
      public void begin(Channels channels) {
          if (running) {
              return;
          }
          lock.lock();
          try {
              this.allocateBuffer2Thread();
              for (ConsumerThread consumerThread : consumerThreads) {
                  consumerThread.start();
              }
              running = true;
          } finally {
              lock.unlock();
          }
      }
  
      @Override
      public boolean isRunning(Channels channels) {
          return running;
      }
  
      private void allocateBuffer2Thread() {
          int channelSize = this.channels.getChannelSize();
          /**
           * if consumerThreads.length < channelSize
           * each consumer will process several channels.
           *
           * if consumerThreads.length == channelSize
           * each consumer will process one channel.
           *
           * if consumerThreads.length > channelSize
           * there will be some threads do nothing.
           */
          for (int channelIndex = 0; channelIndex < channelSize; channelIndex++) {
              int consumerIndex = channelIndex % consumerThreads.length;
              consumerThreads[consumerIndex].addDataSource(channels.getBuffer(channelIndex));
          }
  
      }
  
      @Override
      public void close(Channels channels) {
          lock.lock();
          try {
              this.running = false;
              for (ConsumerThread consumerThread : consumerThreads) {
                  consumerThread.shutdown();
              }
          } finally {
              lock.unlock();
          }
      }
  }
  