content: /**
   * 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.alibaba.jstorm.task.execute.spout;
  
  import java.util.List;
  import java.util.Map;
  
  import com.alibaba.jstorm.task.TaskStatus;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import com.alibaba.jstorm.callback.AsyncLoopThread;
  import com.alibaba.jstorm.client.ConfigExtension;
  import com.alibaba.jstorm.client.spout.CtrlMsgSpout;
  import com.alibaba.jstorm.client.spout.ICtrlMsgSpout;
  import com.alibaba.jstorm.cluster.Common;
  import com.alibaba.jstorm.common.metric.AsmGauge;
  import com.alibaba.jstorm.common.metric.AsmHistogram;
  import com.alibaba.jstorm.common.metric.TimerRatio;
  import com.alibaba.jstorm.daemon.worker.JStormDebugger;
  import com.alibaba.jstorm.daemon.worker.timer.TimerConstants;
  import com.alibaba.jstorm.daemon.worker.timer.TimerTrigger;
  import com.alibaba.jstorm.metric.JStormMetrics;
  import com.alibaba.jstorm.metric.MetricDef;
  import com.alibaba.jstorm.metric.MetricType;
  import com.alibaba.jstorm.metric.MetricUtils;
  import com.alibaba.jstorm.task.Task;
  import com.alibaba.jstorm.task.acker.Acker;
  import com.alibaba.jstorm.task.comm.TupleInfo;
  import com.alibaba.jstorm.task.execute.BaseExecutors;
  import com.alibaba.jstorm.task.master.ctrlevent.TopoMasterCtrlEvent;
  import com.alibaba.jstorm.utils.JStormUtils;
  import com.alibaba.jstorm.utils.Pair;
  import com.alibaba.jstorm.utils.RotatingMap;
  import com.alibaba.jstorm.utils.TimeUtils;
  import com.codahale.metrics.Gauge;
  import com.lmax.disruptor.EventHandler;
  
  import backtype.storm.Config;
  import backtype.storm.spout.ISpout;
  import backtype.storm.spout.SpoutOutputCollector;
  import backtype.storm.tuple.MessageId;
  import backtype.storm.tuple.Tuple;
  import backtype.storm.tuple.TupleExt;
  import backtype.storm.tuple.TupleImplExt;
  
  /**
   * spout executor base class
   *
   * @author yannian/Longda
   */
  public class SpoutExecutors extends BaseExecutors implements EventHandler {
      private static Logger LOG = LoggerFactory.getLogger(SpoutExecutors.class);
  
      protected final Integer maxSpoutPending;
  
      protected backtype.storm.spout.ISpout spout;
      protected RotatingMap<Long, TupleInfo> pending;
  
      protected SpoutOutputCollector outputCollector;
  
      protected AsmHistogram nextTupleTimer;
      protected TimerRatio emptyCpuGauge;
  
      private final String topologyId;
      private final String componentId;
      private int taskId;
  
      protected AsyncLoopThread ackerRunnableThread;
  
      protected boolean isSpoutFullSleep;
  
      protected volatile boolean checkTopologyFinishInit = false;
  
      public SpoutExecutors(Task task) {
          super(task);
  
          this.spout = (ISpout) task.getTaskObj();
  
          int ackerNum = JStormUtils.parseInt(storm_conf.get(Config.TOPOLOGY_ACKER_EXECUTORS));
          if (ackerNum != 0) {
              this.maxSpoutPending = JStormUtils.parseInt(storm_conf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING));
          } else {
              this.maxSpoutPending = null;
          }
          this.topologyId = sysTopologyCtx.getTopologyId();
          this.componentId = sysTopologyCtx.getThisComponentId();
          this.taskId = task.getTaskId();
  
          this.nextTupleTimer = (AsmHistogram) JStormMetrics.registerTaskMetric(MetricUtils.taskMetricName(
                  topologyId, componentId, taskId, MetricDef.EXECUTE_TIME, MetricType.HISTOGRAM), new AsmHistogram());
  
          this.emptyCpuGauge = new TimerRatio();
          JStormMetrics.registerTaskMetric(MetricUtils.taskMetricName(
                          topologyId, componentId, taskId, MetricDef.EMPTY_CPU_RATIO, MetricType.GAUGE),
                  new AsmGauge(emptyCpuGauge));
  
          isSpoutFullSleep = ConfigExtension.isSpoutPendFullSleep(storm_conf);
  
          LOG.info("isSpoutFullSleep:" + isSpoutFullSleep);
  
          mkPending();
  
          JStormMetrics.registerTaskMetric(MetricUtils.taskMetricName(
                  topologyId, componentId, taskId, MetricDef.PENDING_MAP, MetricType.GAUGE), new AsmGauge(
                  new Gauge<Double>() {
                      @Override
                      public Double getValue() {
                          return (double) pending.size();
                      }
                  }));
  
          // collector, in fact it call send_spout_msg
          SpoutCollector collector;
          if (ConfigExtension.isTaskBatchTuple(storm_conf)) {
              collector = new SpoutBatchCollector(task, pending, exeQueue);
          } else {
              collector = new SpoutCollector(task, pending, exeQueue);
          }
  
          this.outputCollector = new SpoutOutputCollector(collector);
          //this task don't continue until it bulid connection with topologyMaster
          Integer topologyMasterId = sysTopologyCtx.getTopologyMasterId();
          List<Integer> localWorkerTasks = sysTopologyCtx.getThisWorkerTasks();
          if (topologyMasterId != 0 && !localWorkerTasks.contains(topologyMasterId)) {
              while (getConnection(topologyMasterId) == null) {
                  JStormUtils.sleepMs(10);
                  LOG.info("this task is still building connection with topology master");
              }
          }
  
          taskHbTrigger.setSpoutOutputCollector(collector);
          taskHbTrigger.register();
  
          LOG.info("Successfully create SpoutExecutors " + idStr);
      }
  
      public void mkPending() {
          // this function will be override by subclass
          throw new RuntimeException("Should override this function");
      }
  
      @Override
      public void init() throws Exception {
          this.spout.open(storm_conf, userTopologyCtx, outputCollector);
          //send the HbMsg to TM, but it don't really finish spout's init.
          taskHbTrigger.sendHbMsg();
          LOG.info("Successfully open SpoutExecutors " + idStr);
      }
  
      public void nextTuple() {
          if (!taskStatus.isRun()) {
              JStormUtils.sleepMs(1);
              return;
          }
  
          // if don't need ack, pending map will be always empty
          if (maxSpoutPending == null || pending.size() < maxSpoutPending) {
              emptyCpuGauge.stop();
  
              long start = nextTupleTimer.getTime();
              try {
                  spout.nextTuple();
              } catch (Throwable e) {
                  error = e;
                  LOG.error("spout execute error ", e);
                  reportError.report(e);
              } finally {
                  nextTupleTimer.updateTime(start);
              }
          } else {
              if (isSpoutFullSleep) {
                  JStormUtils.sleepMs(1);
              }
              emptyCpuGauge.start();
              // just return, no sleep
          }
      }
  
      @Override
      public void run() {
          throw new RuntimeException("Should implement this function");
      }
  
      /**
       * Handle acker message
       *
       * @see com.lmax.disruptor.EventHandler#onEvent(java.lang.Object, long, boolean)
       */
      @Override
      public void onEvent(Object event, long sequence, boolean endOfBatch) throws Exception {
          try {
              if (event == null) {
                  return;
              }
              Runnable runnable = null;
              if (event instanceof Tuple) {
                  if (((TupleExt) event).isBatchTuple()) {
                      List<Object> values = ((Tuple) event).getValues();
                      for (Object value : values) {
                          Pair<MessageId, List<Object>> val = (Pair<MessageId, List<Object>>) value;
                          TupleImplExt tuple = new TupleImplExt(
                                  sysTopologyCtx, val.getSecond(), val.getFirst(), ((TupleImplExt) event));
                          runnable = processTupleEvent(tuple);
                          if (runnable != null) {
                              runnable.run();
                              runnable = null;
                          }
                      }
                  } else {
                      runnable = processTupleEvent((Tuple) event);
                  }
              } else if (event instanceof TimerTrigger.TimerEvent) {
                  processTimerEvent((TimerTrigger.TimerEvent) event);
                  return;
              } else if (event instanceof IAckMsg) {
                  runnable = (Runnable) event;
              } else if (event instanceof Runnable) {
                  runnable = (Runnable) event;
              } else {
  
                  LOG.warn("Receive one unknown event-" + event.toString() + " " + idStr);
                  return;
              }
  
              if (runnable != null)
                  runnable.run();
  
          } catch (Throwable e) {
              if (!taskStatus.isShutdown()) {
                  LOG.info("Unknown exception ", e);
                  reportError.report(e);
              }
          }
      }
  
      private Runnable processTupleEvent(Tuple event) {
          Runnable runnable = null;
          Tuple tuple = event;
          if (event.getSourceStreamId().equals(Common.TOPOLOGY_MASTER_CONTROL_STREAM_ID)) {
              TopoMasterCtrlEvent ctrlEvent = (TopoMasterCtrlEvent) tuple.getValueByField("ctrlEvent");
              if (ctrlEvent.isTransactionEvent()) {
                  if (spout instanceof ICtrlMsgSpout) {
                      runnable = new CtrlMsgSpout((ICtrlMsgSpout) spout, ctrlEvent);
                  }
              } else if (ctrlEvent.isFinishInitEvent()) {
                  LOG.info("spout task-{} received topology finish init operation message", taskId);
                  taskHbTrigger.updateExecutorStatus(TaskStatus.RUN);
                  this.checkTopologyFinishInit = true;
              } else {
                  LOG.warn("Received unexpected control event, {}", ctrlEvent);
              }
          } else if (event.getSourceStreamId().equals(Common.TOPOLOGY_MASTER_REGISTER_METRICS_RESP_STREAM_ID)) {
              this.metricsReporter.updateMetricMeta((Map<String, Long>) tuple.getValue(0));
          } else {
              Object id = tuple.getValue(0);
              Object obj = pending.remove((Long) id);
  
              if (obj == null) {
                  if (JStormDebugger.isDebug(id)) {
                      LOG.info("Pending map no entry:" + id);
                  }
                  runnable = null;
              } else {
                  TupleInfo tupleInfo = (TupleInfo) obj;
  
                  String stream_id = tuple.getSourceStreamId();
  
                  if (stream_id.equals(Acker.ACKER_ACK_STREAM_ID)) {
                      runnable = new AckSpoutMsg(id, spout, tuple, tupleInfo, taskStats);
                  } else if (stream_id.equals(Acker.ACKER_FAIL_STREAM_ID)) {
                      runnable = new FailSpoutMsg(id, spout, tupleInfo, taskStats);
                  } else {
                      LOG.warn("Receive one unknown source Tuple " + idStr);
                      runnable = null;
                  }
              }
  
              taskStats.recv_tuple(tuple.getSourceComponent(), tuple.getSourceStreamId());
          }
          return runnable;
      }
  
      public AsyncLoopThread getAckerRunnableThread() {
          return ackerRunnableThread;
      }
  
      private void processTimerEvent(TimerTrigger.TimerEvent event) {
          switch (event.getOpCode()) {
              case TimerConstants.ROTATING_MAP: {
                  Map<Long, TupleInfo> timeoutMap = pending.rotate();
                  for (java.util.Map.Entry<Long, TupleInfo> entry : timeoutMap.entrySet()) {
                      TupleInfo tupleInfo = entry.getValue();
                      FailSpoutMsg fail = new FailSpoutMsg(entry.getKey(), spout, tupleInfo, taskStats);
                      fail.run();
                  }
                  break;
              }
              case TimerConstants.TASK_HEARTBEAT: {
                  taskHbTrigger.setExeThreadHbTime(TimeUtils.current_time_secs());
                  break;
              }
              default: {
                  LOG.warn("Receive unsupported timer event, opcode=" + event.getOpCode());
                  break;
              }
          }
      }
  
      protected void processControlEvent() {
          Object event = controlQueue.poll();
  
          if (event != null) {
              Runnable runnable = null;
              if (event instanceof TimerTrigger.TimerEvent) {
                  processTimerEvent((TimerTrigger.TimerEvent) event);
              } else if (event instanceof Tuple) {
                  runnable = processTupleEvent((Tuple) event);
              } else {
                  LOG.warn("Received unknown control event, " + event.getClass().getName());
              }
  
              if (runnable != null) {
                  runnable.run();
              }
          }
      }
  
      public Object getOutputCollector() {
          return outputCollector;
      }
  }
  