Function to do some async work on thread class
@param int $timeToWorkOnSomething
@param callable $onComplete
@return AsyncMessage
435157
public function SomeUndefinedLongRunningWork($timeToWorkOnSomething, callable $onComplete = null)
    {
        if ($this->isExternal()) {
            $timeToWorkOnSomething *= 100;
            echo "Start work $timeToWorkOnSomething msec on thread: " . posix_getpid() . PHP_EOL;
            //just a simulation on different time running process
            usleep($timeToWorkOnSomething);
            //not required but return everything you like
            //lets return a message to display on parent
            echo "Completed work $timeToWorkOnSomething msec on thread: " . posix_getpid();
            return "Completed work $timeToWorkOnSomething msec on thread: " . posix_getpid();
        } else {
            return $this->asyncCallOnChild(__FUNCTION__, array($timeToWorkOnSomething), $onComplete, function (AsyncMessage $messga) {
                var_dump($messga->GetResult()->getMessage());
            });
        }
    }
1084422
public function then(callable $onFulfilled = null, callable $onRejected = null, callable $onProgress = null)
    {
        if (null !== $this->result) {
            return $this->result->then($onFulfilled, $onRejected, $onProgress);
        }

        if (null === $this->canceller) {
            return new static($this->resolver($onFulfilled, $onRejected, $onProgress));
        }

        // keep a reference to this promise instance for the static canceller function.
        // see also parentCancellerFunction() for more details.
        $parent = $this;
        ++$parent->requiredCancelRequests;

        return new static(
            $this->resolver($onFulfilled, $onRejected, $onProgress),
            self::parentCancellerFunction($parent)
        );
    }
122234
public function actionComplete(WorkflowTransition $transition)
    {
        $this->MemberID = Member::currentUserID();
        $this->write();
        $this->extend('onActionComplete', $transition);
    }
859360
private void onNextAttemptStarted(CompletionStage<Boolean> futureOutcome) {
    assert executor.inEventLoop();
    state = State.ATTEMPT_IN_PROGRESS;
    futureOutcome
        .whenCompleteAsync(this::onNextAttemptCompleted, executor)
        .exceptionally(UncaughtExceptions::log);
  }
1051598
@Restricted(NoExternalUse.class)
    public synchronized final void maybeComplete() {
        assert this.executor != null;
        if (result != null) {
            executor.completedAsynchronous(result != NULL ? result : null);
            result = null;
        }
    }
414436
public function onComplete(CompleteEvent $event)
    {
        $code = (string) $event->getResponse()->getStatusCode();
        // Throw an exception for an unsuccessful response
        if ($code[0] === '4' || $code[0] === '5') {
            throw RequestException::create($event->getRequest(), $event->getResponse());
        }
    }
1154271
private void waitForCompletion() {
        for (ListenableFuture<?> future : pendingTasks.values()) {
            try {
                future.get();
            } catch (InterruptedException | ExecutionException e) {
                LOG.error("[" + this + "] Error waiting for writes to complete: " + e.getMessage());
            }
        }
    }
547657
public void complete(NetworkConnection vc, IOReadRequestContext rctx)
   {	
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc, "complete", new Object[] {vc, rctx});
      if (thisConnection.isLoggingIOEvents()) thisConnection.getConnectionEventRecorder().logDebug("complete method invoked on read context "+System.identityHashCode(rctx));

      // First update the invocation count. If we are being called back on the same thread as
      // last time, increment the counter. Otherwise, start counting from 1 again.
      synchronized (invocationCountLock)
      {
         if (lastInvokedOnThread == Thread.currentThread())
         {
            invocationCount++;
         }
         else
         {
            invocationCount = 1;
            lastInvokedOnThread = Thread.currentThread();
         }
      }

      try
      {
         synchronized(this)
         {
            boolean done = false;
            do
            {
               done = true;
               WsByteBuffer contextBuffer = rctx.getBuffer();

               contextBuffer.flip();

               // Notify PMI that read has completed.
               if (conversation.getConversationType() == Conversation.CLIENT)
               {
                  xmitParser.setType(Conversation.CLIENT);
               }
               else if (conversation.getConversationType() == Conversation.ME)
               {
                  xmitParser.setType(Conversation.ME);
               }

               if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) JFapUtils.debugTraceWsByteBuffer(this, tc, contextBuffer, 16, "data received");

               xmitParser.parse(contextBuffer);

         		if ((rctx != null) && (!receivePhysicalCloseRequest))
         		{
                  // Calculate the amount of time before the request times out
                  // This is the mechanism by which we implement heartbeat
                  // intervals and time outs.
                  int timeout;
                  if (awaitingHeartbeatResponse.isSet())
                  {
                     // We can only reach this point in the code if we have
                     // made a heartbeat request but this callback was driven
                     // because of a non-heartbeat response.  In previous versions
                     // of the code we were careful to calculate the remaining time and
                     // make another read request.  This caused timeouts ala defect
                     // 363463.  So, now the code is more generous and resets its
                     // timer to the full heartbeat timeout value every time any
                     // non-heartbeat trasmission is received from our peer.  This isn't
                     // unreasonable as the fact we are receiving data indicates that the
                     // peer is probably still healthy.
                     timeout = currentHeartbeatTimeout*1000;
                     if (timeout < 1) timeout = 1;
                  }
                  else
                  {
                     // We are not awaiting a heartbeat response, use the
                     // heartbeat interval as our timeout.
                     timeout = thisConnection.getHeartbeatInterval()*1000;
                  }

                  if (timeout > 0)
                  {
                     if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "setting heartbeat timeout to: "+timeout+" milliseconds");
                  }
                  else
                  {
                     if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "not using a heartbeat timeout");
                     timeout = IOReadRequestContext.NO_TIMEOUT;
                  }

                  boolean closing = false;
                  synchronized(connectionClosingLock)
                  {
                     closing = connectionClosing;
                  }

                  //If the connection is closing/closed our buffers will have been released, so make sure we don't use them again.
                  if (!closing)
                  {
                     // Crude way to ensure we end up with the right sized read buffer.
                     if (isFirstCompleteInvocation)
                     {
                        int readBufferSize =
                           Integer.parseInt(RuntimeInfo.getProperty("com.ibm.ws.sib.jfapchannel.DEFAULT_READ_BUFFER_SIZE", ""+JFapChannelConstants.DEFAULT_READ_BUFFER_SIZE));
                        if (!contextBuffer.isDirect() || (contextBuffer.capacity() < readBufferSize))
                        {
                           //Make sure we release the other buffer so we don't leak memory.
                           contextBuffer.release();

                           contextBuffer = WsByteBufferPool.getInstance().allocateDirect(readBufferSize);
                           rctx.setBuffer(contextBuffer);
                        }
                        isFirstCompleteInvocation = false;
                     }

                     contextBuffer.clear();

                     // Decide whether to explictly request a thread switch. We'll do this if we
                     // have been recursively called more than MAX_INVOCATIONS_BEFORE_THREAD_SWITCH
                     boolean forceQueue = false;
                     synchronized (invocationCountLock)
                     {
                        if (invocationCount > MAX_INVOCATIONS_BEFORE_THREAD_SWITCH)
                        {
                           forceQueue = true;
                           lastInvokedOnThread = null;
                        }
                     }

                     if (thisConnection.isLoggingIOEvents()) thisConnection.getConnectionEventRecorder().logDebug("invoking readCtx.read() on context "+System.identityHashCode(rctx)+" with a timeout of "+timeout);
                     done = (rctx.read(1, this, forceQueue, timeout) == null);
                   
                  }
         		}
            }
            while(!done);
         }
      }
      catch(Error error)
      {
         FFDCFilter.processException
         (error, "com.ibm.ws.sib.jfapchannel.impl.ConnectionReadCompletedCallback", JFapChannelConstants.CONNREADCOMPCALLBACK_COMPLETE_01, thisConnection.getDiagnostics(true));
         if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled()) SibTr.exception(this, tc, error);

         // It might appear slightly odd for this code to catch Error (especially since the JDK docs say
         // that Error means that something has gone so badly wrong that you should abandon all hope).
         // This code makes one final stab at putting out some diagnostics about what happened (if we
         // propagate the Error up to the TCP Channel, it is sometimes lost) and closing down the
         // connection.  I figured that we might as well try to do something - as we can hardly make
         // things worse... (famous last words)

         thisConnection.invalidate(false, error, "Error caught in ConnectionReadCompletedCallback.complete()");

         // Re-throw the error to ensure that it causes the maximum devastation.
         // The JVM is probably very ill if an Error is thrown so attempt no recovery.
         throw error;
      }
      catch(RuntimeException runtimeException)
      {
         FFDCFilter.processException
         (runtimeException, "com.ibm.ws.sib.jfapchannel.impl.ConnectionReadCompletedCallback", JFapChannelConstants.CONNREADCOMPCALLBACK_COMPLETE_02, thisConnection.getDiagnostics(true));
         if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled()) SibTr.exception(this, tc, runtimeException);

         // We can reasonably try to recover from a runtime exception by invalidating the associated
         // connection.  This should drive the underlying TCP/IP socket to be closed.
         thisConnection.invalidate(false, runtimeException, "RuntimeException caught in ConnectionReadCompletedCallback.complete");

         // Don't throw the RuntimeException on as we risk blowing away part of the TCP channel.
      }

		if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "complete");
	}
895171
protected Function<Code, Timer> asTimerFunction(final Supplier<Timer.Builder> timerTemplate) {
        final Map<Code, Timer> cache = new EnumMap<>(Code.class);
        final Function<Code, Timer> creator = code -> timerTemplate.get()
                .tag(TAG_STATUS_CODE, code.name())
                .register(this.registry);
        final Function<Code, Timer> cacheResolver = code -> cache.computeIfAbsent(code, creator);
        // Eager initialize
        for (final Code code : this.eagerInitializedCodes) {
            cacheResolver.apply(code);
        }
        return cacheResolver;
    }
1220860
public void onSuccessfulWrite(long startTimeNanos) {
    Instrumented.updateTimer(this.dataWriterTimer, System.nanoTime() - startTimeNanos, TimeUnit.NANOSECONDS);
    Instrumented.markMeter(this.successfulWritesMeter);
  }
431794
public static <T> ObservableTransformer<T,T> doOnCompleteFx(Action onCompleted) {
        return obs -> obs.doOnComplete(() -> runOnFx(onCompleted));
    }
419795
@Override
    public synchronized boolean send(final Object message,
                                     final boolean shouldClose,
                                     final OnComplete onComplete) throws IOException {
        if (!isOpen()) {
            return false;
        }

        this.sendQueued = true;

        byte[] data;
        if (message == null) {
            data = null;
        } else if (message instanceof String) {
            data = ((String)message).getBytes(getResponseCharset());
        } else if (message instanceof byte[]) {
            data = (byte[])message;
        } else {
            throw WebsocketUtil.wrongMessageType(message.getClass());
        }

        enqueue(new PendingSend(data, shouldClose, onComplete));

        return true;
    }
94401
public function wait($terminationCallback = null, $addToCompletedQueue = true)
    {
        $lastProcessesRunningCount = $currentRunningProcessesCount = $this->countRunning();
        while ($currentRunningProcessesCount > 0) {
            $currentRunningProcessesCount = $this->countRunning();
            if ($lastProcessesRunningCount !== $currentRunningProcessesCount) {
                $lastProcessesRunningCount = $currentRunningProcessesCount;
                $this->cleanUP($addToCompletedQueue);
                if (null !== $terminationCallback) {
                    call_user_func($terminationCallback);
                }
            }
            usleep(1000);
        }

        return true;
    }
179267
public function addTimeEvent(TimeEvent $timeEvent)
    {
        $this->timeEvents[] = $timeEvent;
        $this->eventHandler->timeEventAdded($this, $timeEvent);
    }
235818
public function onComplete(CompleteEvent $event)
    {
        $request = $event->getRequest();
        $response = $event->getResponse();

        // Cache the response if it can be cached and isn't already
        if ($request->getConfig()->get('cache_lookup') === 'MISS'
            && call_user_func($this->canCache, $request)
            && Utils::canCacheResponse($response)
        ) {
            // Store the date when the response was cached
            $response->setHeader('X-Guzzle-Cache-Date', gmdate('D, d M Y H:i:s T', time()));
            $this->storage->cache($request, $response);
        }

        $this->addResponseHeaders($request, $response);
    }
306986
public static <T> CompletableFuture<T> orTimeout(CompletableFuture<T> future, long timeout, TimeUnit timeUnit) {
		return orTimeout(future, timeout, timeUnit, Executors.directExecutor());
	}
210057
protected Object doInvokeMethod(String methodName, Object name, Object args) {
        super.doInvokeMethod(methodName, name, args);


        // return the completed node
        return lastCompletedNode;
    }