Retrieves all the event rows matching a single
{@link com.twitter.hraven.Flow}.
@param flowKey
@return
10469
public List<FlowEvent> getFlowEvents(FlowKey flowKey) throws IOException {
    byte[] startKey =
        Bytes.add(flowKeyConverter.toBytes(flowKey), Constants.SEP_BYTES);
    Scan scan = new Scan(startKey);
    scan.setFilter(new WhileMatchFilter(new PrefixFilter(startKey)));

    List<FlowEvent> results = new ArrayList<FlowEvent>();
    ResultScanner scanner = null;
    Table eventTable = null;
    try {
      eventTable = hbaseConnection
          .getTable(TableName.valueOf(Constants.FLOW_EVENT_TABLE));
      scanner = eventTable.getScanner(scan);
      for (Result r : scanner) {
        FlowEvent event = createEventFromResult(r);
        if (event != null) {
          results.add(event);
        }
      }
    } finally {
      try {
        if (scanner != null) {
          scanner.close();
        }
      } finally {
        if (eventTable != null) {
          eventTable.close();
        }
      }
    }
    return results;
  }
1030853
public java.util.Map<String, String> getEvents() {
        if (events == null) {
            events = new com.amazonaws.internal.SdkInternalMap<String, String>();
        }
        return events;
    }
413636
public static EventType getEventType(Key key)
    {
        Key ek = findAncestor(Repository.EVENTTYPE, key);
        return EventType.values()[(int)ek.getId()-1];
    }
1016381
public List<Integer> getQueuedFlowIds() {
    final List<Integer> allIds = new ArrayList<>();
    try {
      getExecutionIdsHelper(allIds, this.executorLoader.fetchQueuedFlows());
    } catch (final ExecutorManagerException e) {
      this.logger.error("Failed to get queued flow ids.", e);
    }
    return allIds;
  }
623088
def ReadAllFlowObjects(
      self,
      client_id = None,
      min_create_time = None,
      max_create_time = None,
      include_child_flows = True,
  ):
    """"""
    res = []
    for flow in itervalues(self.flows):
      if ((client_id is None or flow.client_id == client_id) and
          (min_create_time is None or flow.create_time >= min_create_time) and
          (max_create_time is None or flow.create_time <= max_create_time) and
          (include_child_flows or not flow.parent_flow_id)):
        res.append(flow.Copy())
    return res
653201
@Override
  public List<ExecutableFlow> getRunningFlows() {
    final ArrayList<ExecutableFlow> flows = new ArrayList<>();
    getActiveFlowHelper(flows, this.queuedFlows.getAllEntries());
    getActiveFlowHelper(flows, this.runningExecutions.get().values());
    return flows;
  }
1237366
public FlowStatus getFlowStatus(FlowStatusId flowStatusId)
      throws RemoteInvocationException {
    LOG.debug("getFlowConfig with groupName " + flowStatusId.getFlowGroup() + " flowName " +
        flowStatusId.getFlowName());

    GetRequest<FlowStatus> getRequest = _flowstatusesRequestBuilders.get()
        .id(new ComplexResourceKey<>(flowStatusId, new EmptyRecord())).build();

    Response<FlowStatus> response =
        _restClient.get().sendRequest(getRequest).getResponse();
    return response.getEntity();
  }
295669
private Map<Optional<Executor>, List<ExecutableFlow>> getFlowToExecutorMap() {
    final HashMap<Optional<Executor>, List<ExecutableFlow>> exFlowMap = new HashMap<>();
    try {
      for (final Pair<ExecutionReference, ExecutableFlow> runningFlow : this
          .executorLoader.fetchActiveFlows().values()) {
        final Optional<Executor> executor = runningFlow.getFirst().getExecutor();
        List<ExecutableFlow> flows = exFlowMap.get(executor);
        if (flows == null) {
          flows = new ArrayList<>();
          exFlowMap.put(executor, flows);
        }
        flows.add(runningFlow.getSecond());
      }
    } catch (final ExecutorManagerException e) {
      logger.error("Failed to get flow to executor map");
    }
    return exFlowMap;
  }
172012
public List<EventData> parse(Long pipelineId, List<Entry> datas) throws SelectException {
        List<EventData> eventDatas = new ArrayList<EventData>();
        Pipeline pipeline = configClientService.findPipeline(pipelineId);
        List<Entry> transactionDataBuffer = new ArrayList<Entry>();
        // hz为主站点，us->hz的数据，需要回环同步会us。并且需要开启回环补救算法
        PipelineParameter pipelineParameter = pipeline.getParameters();
        boolean enableLoopbackRemedy = pipelineParameter.isEnableRemedy() && pipelineParameter.isHome()
                                       && pipelineParameter.getRemedyAlgorithm().isLoopback();
        boolean isLoopback = false;
        boolean needLoopback = false; // 判断是否属于需要loopback处理的类型，只处理正常otter同步产生的回环数据，因为会有业务方手工屏蔽同步的接口，避免回环

        long now = new Date().getTime();
        try {
            for (Entry entry : datas) {
                switch (entry.getEntryType()) {
                    case TRANSACTIONBEGIN:
                        isLoopback = false;
                        break;
                    case ROWDATA:
                        String tableName = entry.getHeader().getTableName();
                        // 判断是否是回环表retl_mark
                        boolean isMarkTable = tableName.equalsIgnoreCase(pipeline.getParameters().getSystemMarkTable());
                        if (isMarkTable) {
                            RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
                            if (!rowChange.getIsDdl()) {
                                int loopback = 0;
                                if (rowChange.getRowDatasCount() > 0) {
                                    loopback = checkLoopback(pipeline, rowChange.getRowDatas(0));
                                }
                                if (loopback == 2) {
                                    needLoopback |= true; // 只处理正常同步产生的回环数据
                                }

                                isLoopback |= loopback > 0;
                            }
                        }

                        // 检查下otter3.0的回环表，对应的schmea会比较随意，所以不做比较
                        boolean isCompatibleLoopback = tableName.equalsIgnoreCase(compatibleMarkTable);

                        if (isCompatibleLoopback) {
                            RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
                            if (!rowChange.getIsDdl()) {
                                int loopback = 0;
                                if (rowChange.getRowDatasCount() > 0) {
                                    loopback = checkCompatibleLoopback(pipeline, rowChange.getRowDatas(0));
                                }
                                if (loopback == 2) {
                                    needLoopback |= true; // 只处理正常同步产生的回环数据
                                }
                                isLoopback |= loopback > 0;
                            }
                        }

                        if ((!isLoopback || (enableLoopbackRemedy && needLoopback)) && !isMarkTable
                            && !isCompatibleLoopback) {
                            transactionDataBuffer.add(entry);
                        }
                        break;
                    case TRANSACTIONEND:
                        if (!isLoopback || (enableLoopbackRemedy && needLoopback)) {
                            // 添加数据解析
                            for (Entry bufferEntry : transactionDataBuffer) {
                                List<EventData> parseDatas = internParse(pipeline, bufferEntry);
                                if (CollectionUtils.isEmpty(parseDatas)) {// 可能为空，针对ddl返回时就为null
                                    continue;
                                }

                                // 初步计算一下事件大小
                                long totalSize = bufferEntry.getHeader().getEventLength();
                                long eachSize = totalSize / parseDatas.size();
                                for (EventData eventData : parseDatas) {
                                    if (eventData == null) {
                                        continue;
                                    }

                                    eventData.setSize(eachSize);// 记录一下大小
                                    if (needLoopback) {// 针对需要回环同步的
                                        // 如果延迟超过指定的阀值，则设置为需要反查db
                                        if (now - eventData.getExecuteTime() > 1000 * pipeline.getParameters()
                                            .getRemedyDelayThresoldForMedia()) {
                                            eventData.setSyncConsistency(SyncConsistency.MEDIA);
                                        } else {
                                            eventData.setSyncConsistency(SyncConsistency.BASE);
                                        }
                                        eventData.setRemedy(true);
                                    }
                                    eventDatas.add(eventData);
                                }
                            }
                        }

                        isLoopback = false;
                        needLoopback = false;
                        transactionDataBuffer.clear();
                        break;
                    default:
                        break;
                }
            }

            // 添加最后一次的数据，可能没有TRANSACTIONEND
            if (!isLoopback || (enableLoopbackRemedy && needLoopback)) {
                // 添加数据解析
                for (Entry bufferEntry : transactionDataBuffer) {
                    List<EventData> parseDatas = internParse(pipeline, bufferEntry);
                    if (CollectionUtils.isEmpty(parseDatas)) {// 可能为空，针对ddl返回时就为null
                        continue;
                    }

                    // 初步计算一下事件大小
                    long totalSize = bufferEntry.getHeader().getEventLength();
                    long eachSize = totalSize / parseDatas.size();
                    for (EventData eventData : parseDatas) {
                        if (eventData == null) {
                            continue;
                        }

                        eventData.setSize(eachSize);// 记录一下大小
                        if (needLoopback) {// 针对需要回环同步的
                            // 如果延迟超过指定的阀值，则设置为需要反查db
                            if (now - eventData.getExecuteTime() > 1000 * pipeline.getParameters()
                                .getRemedyDelayThresoldForMedia()) {
                                eventData.setSyncConsistency(SyncConsistency.MEDIA);
                            } else {
                                eventData.setSyncConsistency(SyncConsistency.BASE);
                            }
                        }
                        eventDatas.add(eventData);
                    }
                }
            }
        } catch (Exception e) {
            throw new SelectException(e);
        }

        return eventDatas;
    }
1042308
@Override
  public Set<FlowEdge> getEdges(String nodeId) {
    try {
      rwLock.readLock().lock();
      DataNode dataNode = this.dataNodeMap.getOrDefault(nodeId, null);
      return getEdges(dataNode);
    } finally {
      rwLock.readLock().unlock();
    }
  }
954350
public Flow[] getFlows()
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
    {
      SibTr.entry(tc, "getFlows");
    }
    Flow[] clonedFlows = flows.clone();
    
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
    {    
      SibTr.exit(tc, "getFlows", clonedFlows);
    }    
    return clonedFlows;
  }
202410
public List<FacesConfigFlowDefinitionType<WebFacesConfigDescriptor>> getAllFlowDefinition()
   {
      List<FacesConfigFlowDefinitionType<WebFacesConfigDescriptor>> list = new ArrayList<FacesConfigFlowDefinitionType<WebFacesConfigDescriptor>>();
      List<Node> nodeList = model.get("flow-definition");
      for(Node node: nodeList)
      {
         FacesConfigFlowDefinitionType<WebFacesConfigDescriptor>  type = new FacesConfigFlowDefinitionTypeImpl<WebFacesConfigDescriptor>(this, "flow-definition", model, node);
         list.add(type);
      }
      return list;
   }
1117305
def ReadChildFlowObjects(self, client_id, flow_id):
    """"""
    res = []
    for flow in itervalues(self.flows):
      if flow.client_id == client_id and flow.parent_flow_id == flow_id:
        res.append(flow)
    return res
592124
public List<FacesConfigFlowDefinitionFlowCallType<FacesConfigFlowDefinitionType<T>>> getAllFlowCall()
   {
      List<FacesConfigFlowDefinitionFlowCallType<FacesConfigFlowDefinitionType<T>>> list = new ArrayList<FacesConfigFlowDefinitionFlowCallType<FacesConfigFlowDefinitionType<T>>>();
      List<Node> nodeList = childNode.get("flow-call");
      for(Node node: nodeList)
      {
         FacesConfigFlowDefinitionFlowCallType<FacesConfigFlowDefinitionType<T>>  type = new FacesConfigFlowDefinitionFlowCallTypeImpl<FacesConfigFlowDefinitionType<T>>(this, "flow-call", childNode, node);
         list.add(type);
      }
      return list;
   }
399578
public java.util.List<EventTopic> getEventTopics() {
        if (eventTopics == null) {
            eventTopics = new com.amazonaws.internal.SdkInternalList<EventTopic>();
        }
        return eventTopics;
    }
795172
public java.util.List<String> getJobFlowIds() {
        if (jobFlowIds == null) {
            jobFlowIds = new com.amazonaws.internal.SdkInternalList<String>();
        }
        return jobFlowIds;
    }
1248024
public Map<String, Process> getRuleFlows() {
        ProcessPackage rtp = (ProcessPackage) getResourceTypePackages().get(ResourceType.BPMN2);
        return rtp == null? Collections.emptyMap() : rtp.getRuleFlows();
    }