Given two SegmentCompletionListener, return a SegmentCompletionListener that
executes both in sequence, even if the first throws an exception, and if both
throw exceptions, add any exceptions thrown by the second as suppressed
exceptions of the first.
5305
protected static Consumer<Supplier<PrimitiveIterator.OfInt>> composeWithExceptions(Consumer<Supplier<PrimitiveIterator.OfInt>> a,
         Consumer<Supplier<PrimitiveIterator.OfInt>> b) {
      return (segments) -> {
         try {
            a.accept(segments);
         }
         catch (Throwable e1) {
            try {
               b.accept(segments);
            }
            catch (Throwable e2) {
               try {
                  e1.addSuppressed(e2);
               } catch (Throwable ignore) {}
            }
            throw e1;
         }
         b.accept(segments);
      };
   }
1188667
protected Future<T> addToListeners(
    final GenericCompletionListener<? extends Future<T>> listener) {
    if (listener == null) {
      throw new IllegalArgumentException("The listener can't be null.");
    }

    synchronized(this) {
      listeners.add(listener);
    }

    if(isDone()) {
      notifyListeners();
    }

    return this;
  }
867452
public void recomputeSegmentKeys() throws IOException {
    if (!closed && segmentRegistration != null) {
      if (segmentRegistration.equals(SEGMENT_SORT_ASC)
          || segmentRegistration.equals(SEGMENT_SORT_DESC)
          || segmentRegistration.equals(SEGMENT_BOUNDARY_ASC)
          || segmentRegistration.equals(SEGMENT_BOUNDARY_DESC)) {

        if (segmentRegistration.equals(SEGMENT_SORT_ASC)
            || segmentRegistration.equals(SEGMENT_SORT_DESC)) {
          segmentKeys.clear();
          // recompute boundaries
          for (Entry<String, Map<String, T1>> entry : segmentKeyValueList
              .entrySet()) {
            T1 tmpSegmentValueBoundary = boundaryForSegment(entry.getKey());
            segmentValuesBoundary.put(entry.getKey(), tmpSegmentValueBoundary);
          }
          // compute adjusted boundaries and compute keys
          for (Entry<String, Map<String, T1>> entry : segmentKeyValueList
              .entrySet()) {
            this.segmentName = entry.getKey();
            Map<String, T1> keyValueList = entry.getValue();
            T1 tmpSegmentValueBoundaryForComputing = boundaryForSegmentComputing(
                entry.getKey());
            for (Entry<String, T1> subEntry : keyValueList.entrySet()) {
              if (tmpSegmentValueBoundaryForComputing == null
                  || compareWithBoundary(subEntry.getValue(),
                      tmpSegmentValueBoundaryForComputing)) {
                if (!segmentKeys.contains(subEntry.getKey())) {
                  segmentKeys.add(subEntry.getKey());
                }
              }
            }
          }
        }

        Map<String, T1> keyValueList;
        Set<String> recomputeKeyList;
        segmentRecomputeKeyList = new LinkedHashMap<>();
        for (String key : segmentKeys) {
          for (Entry<String, Map<String, T1>> entry : segmentKeyValueList
              .entrySet()) {
            keyValueList = entry.getValue();
            if (!keyValueList.containsKey(key)) {
              if (!segmentRecomputeKeyList.containsKey(entry.getKey())) {
                recomputeKeyList = new HashSet<>();
                segmentRecomputeKeyList.put(entry.getKey(), recomputeKeyList);
              } else {
                recomputeKeyList = segmentRecomputeKeyList.get(entry.getKey());
              }
              recomputeKeyList.add(key);
            }
          }
        }
        this.segmentName = null;
      } else {
        throw new IOException(
            "not for segmentRegistration " + segmentRegistration);
      }
    } else {
      throw new IOException("already closed or no segmentRegistration ("
          + segmentRegistration + ")");
    }
  }
1363541
synchronized void onComplete(@javax.annotation.Nullable Throwable error) {
    if (isCompleted()) {
      return;
    }
    currentConfigObserver = null;
    // TODO(songya): add Runnable
    Status status;
    if (error == null) {
      status = Status.OK;
    } else if (error instanceof StatusRuntimeException) {
      status = ((StatusRuntimeException) error).getStatus();
    } else {
      status = Status.UNKNOWN;
    }
    terminateStatus = status;
  }
793955
private void onHandlerEntry() {
        if (enabled) {
            Type exceptionType = handlers.get(handlerPendingInstruction);

            // Clear the pending instruction flag
            handlerPendingInstruction = null;

            // Filter the interested down to this exception
            Set<ProbeListener> filtered = new HashSet<ProbeListener>();
            for (ProbeListener listener : enabledListeners) {
                ListenerConfiguration config = listener.getListenerConfiguration();
                ProbeFilter filter = config.getTransformerData(EXCEPTION_FILTER_KEY);
                if (filter.isBasicFilter() && filter.basicClassNameMatches(exceptionType.getClassName())) {
                    filtered.add(listener);
                } else if (filter.isAdvancedFilter()) {
                    Class<?> clazz = getOwningClass(exceptionType.getInternalName());
                    if (clazz != null && filter.matches(clazz)) {
                        filtered.add(listener);
                    }
                }
            }

            if (filtered.isEmpty()) {
                return;
            }

            String key = createKey(exceptionType);
            ProbeImpl probe = getProbe(key);
            long probeId = probe.getIdentifier();

            setProbeInProgress(true);
            visitInsn(DUP); // throwable throwable
            visitLdcInsn(Long.valueOf(probeId)); // throwable throwable long1 long2
            visitInsn(DUP2_X1); // throwable long1 long2 throwable long1 long2
            visitInsn(POP2); // throwable long1 long2 throwable
            if (isStatic()) {
                visitInsn(ACONST_NULL); // throwable long1 long2 throwable this
            } else {
                visitVarInsn(ALOAD, 0); // throwable long1 long2 throwable this
            }
            visitInsn(SWAP); // throwable long1 long2 this throwable
            visitInsn(ACONST_NULL); // throwable long1 long2 this throwable null
            visitInsn(SWAP); // throwable long1 long2 this null throwable
            visitFireProbeInvocation(); // throwable
            setProbeInProgress(false);

            setProbeListeners(probe, filtered);
        }
    }
804174
public static void postTaskSuccess(final TaskCompletionListener completionListener) {
        new Handler(Looper.getMainLooper()).post(new Runnable() {
            @Override
            public void run() {
                if (null != completionListener) {
                    completionListener.onSuccess();
                }
            }
        });
    }
317016
@SuppressWarnings("unchecked")
    private void prepareListeners() {
        if (preparedListeners != null && preparedUnspecifiedListeners != null) {
            // Already created, skip
            return;
        }
        preparedListeners = new ConcurrentHashMap<>();
        Stream<Class<? extends GloballyAttachableListener>> eventTypes = Stream.concat(
                listeners.keySet().stream(),
                Stream.concat(listenerSuppliers.keySet().stream(),
                        listenerFunctions.keySet().stream())
        ).distinct();
        eventTypes.forEach(type -> {
            ArrayList<Function<DiscordApi, GloballyAttachableListener>> typeListenerFunctions = new ArrayList<>();
            listeners.getOrDefault(type, Collections.emptyList()).forEach(
                    listener -> typeListenerFunctions.add(api -> listener)
            );
            listenerSuppliers.getOrDefault(type, Collections.emptyList()).forEach(
                    supplier -> typeListenerFunctions.add(api -> supplier.get())
            );
            listenerFunctions.getOrDefault(type, Collections.emptyList()).forEach(
                    function -> typeListenerFunctions.add((Function<DiscordApi, GloballyAttachableListener>) function)
            );
            preparedListeners.put(type, typeListenerFunctions);
        });
        // Unspecified Listeners
        preparedUnspecifiedListeners = new CopyOnWriteArrayList<>(unspecifiedListenerFunctions);
        unspecifiedListenerSuppliers.forEach(supplier -> preparedUnspecifiedListeners.add((api) -> supplier.get()));
        unspecifiedListeners.forEach(listener -> preparedUnspecifiedListeners.add((api) -> listener));
    }
1190616
public synchronized Iterator receiveListenerIterator()
	{
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc, "receiveListenerIterator");
		
      final LinkedList<ReceiveListener>linkedList = new LinkedList<ReceiveListener>();
		final Iterator iterator = table.values().iterator();

		while(iterator.hasNext())
		{
			final RequestIdTableEntry tableEntry = (RequestIdTableEntry)iterator.next();
         if (tableEntry.receiveListener != null) linkedList.add(tableEntry.receiveListener);
		}

      final Iterator result = linkedList.iterator();
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "receiveListenerIterator", result);   	
		return result;
	}
1087847
@Override public boolean onExceptionalCompletion(Throwable ex, CountedCompleter caller){
      Futures fs = new Futures();
      if( _job != null ) {
        UKV.remove(_job.destination_key,fs);
        UKV.remove(_job._progress,fs);
        // Find & remove all partially-built output vecs & chunks
        if( _job._mfpt != null ) _job._mfpt.onExceptionCleanup(fs);
      }
      // Assume the input is corrupt - or already partially deleted after
      // parsing.  Nuke it all - no partial Vecs lying around.
      for( Key k : _keys ) UKV.remove(k,fs);
      fs.blockForPending();
      // As soon as the job is canceled, threads blocking on the job will
      // wake up.  Better have all cleanup done first!
      if( _job != null ) _job.cancel(ex);
      return true;
    }
992131
public RetryPolicy<R> onAbort(CheckedConsumer<? extends ExecutionCompletedEvent<R>> listener) {
    abortListener = EventListener.of(Assert.notNull(listener, "listener"));
    return this;
  }
409819
private void postGetGroupMembersError(final GetGroupMembersCompletionListener completionListener, final String errorMessage) {
        new Handler(Looper.getMainLooper()).post(new Runnable() {
            @Override
            public void run() {
                if (null != completionListener) {
                    completionListener.onError(errorMessage);
                }
            }
        });
    }
1293289
final int recordExceptionalCompletion(Throwable ex) {
        int s;
        if ((s = status) >= 0) {
            int h = System.identityHashCode(this);
            final ReentrantLock lock = exceptionTableLock;
            lock.lock();
            try {
                expungeStaleExceptions();
                ExceptionNode[] t = exceptionTable;
                int i = h & (t.length - 1);
                for (ExceptionNode e = t[i]; ; e = e.next) {
                    if (e == null) {
                        t[i] = new ExceptionNode(this, ex, t[i]);
                        break;
                    }
                    if (e.get() == this) // already present
                        break;
                }
            } finally {
                lock.unlock();
            }
            s = setCompletion(EXCEPTIONAL);
        }
        return s;
    }
316644
@Override
	public void setException(@NotNull Throwable e) {
		assert !isComplete();
		exception = e;
		completeExceptionally(e);
	}
862455
private void postConnectError(final ConnectCompletionListener completionListener, final String errorMessage) {
        new Handler(Looper.getMainLooper()).post(new Runnable() {
            @Override
            public void run() {
                if (null != completionListener) {
                    completionListener.onError(errorMessage);
                }
            }
        });
    }
144673
private ReceiveListenerErrorOccurredInvocation allocateErrorOccurredInvocation(Connection connection,
                                                                   ReceiveListener listener,
                                                                   SIConnectionLostException exception,
                                                                   int segmentType,
                                                                   int requestNumber,
                                                                   int priority,
                                                                   Conversation conversation)
   {
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc, "allocateErrorOccurredInvocation", new Object[] {connection, listener, exception, ""+segmentType, ""+requestNumber, ""+priority, conversation});
      ReceiveListenerErrorOccurredInvocation retInvocation = (ReceiveListenerErrorOccurredInvocation)receiveListenerErrorOccurredInvocationPool.remove();
      if (retInvocation == null)
      {
         retInvocation = new ReceiveListenerErrorOccurredInvocation(connection,
                                                                    listener,
                                                                    exception,
                                                                    segmentType,
                                                                    requestNumber,
                                                                    priority,
                                                                    conversation,
                                                                    receiveListenerErrorOccurredInvocationPool);
      }
      else
      {
         retInvocation.reset(connection,
                             listener,
                             exception,
                             segmentType,
                             requestNumber,
                             priority,
                             conversation);
      }

      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "allocateErrorOccurredInvocation", retInvocation);
      return retInvocation;
   }
958617
private int setExceptionalCompletion(Throwable ex) {
        int s = recordExceptionalCompletion(ex);
        if ((s & DONE_MASK) == EXCEPTIONAL)
            internalPropagateException(ex);
        return s;
    }
903304
public void queueErrorOccurredInvocation(Connection connection,
                                               ReceiveListener listener,
                                               SIConnectionLostException exception,
                                               int segmentType,
                                               int requestNumber,
                                               int priority,
                                               Conversation conversation)
   {
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc, "queueErrorOccurredInvocation", new Object[] {connection, listener, exception, ""+segmentType, ""+requestNumber, ""+priority, conversation});

      AbstractInvocation invocation =
         allocateErrorOccurredInvocation(connection,
                                         listener,
                                         exception,
                                         segmentType,
                                         requestNumber,
                                         priority,
                                         conversation);
      queueInvocationCommon(invocation, (ConversationImpl)conversation);
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "queueErrorOccurredInvocation");
   }