afterCompleted will need to handle COMPLETED event by sending message to the peer
358622
public void afterCompleted(Event event) {
		ChaincodeMessage message = messageHelper(event);
		logger.debug(String.format("[%s]sending COMPLETED to validator for tid", shortID(message)));
		try {
			serialSend(message);
		} catch (Exception e) {
			event.cancel(new Exception("send COMPLETED failed %s", e));
		}
	}
695306
private void handlePeerMessage(PeerMessage message) {
		// Ignore our own updates
		Peer peer = PeerUtils.fromMessage(message, Status.ACTIVE);
		if(peer.getId().equals(self)) {
			return;
		}

		// Add or update the peer
		synchronized(this) {
			if(!peers.containsKey(peer.getId())) {
				add(peer);
			}
			else {
				update(peer);
			}
		}
	}
444014
@Override
  public void onCompleted() {
    RpcUtils.streamingRPCAndLog(LOG, new RpcUtils.StreamingRpcCallable<OpenLocalBlockResponse>() {
      @Override
      public OpenLocalBlockResponse call() throws Exception {
        if (mLockId != BlockLockManager.INVALID_LOCK_ID) {
          mWorker.unlockBlock(mLockId);
          mLockId = BlockLockManager.INVALID_LOCK_ID;
        } else if (mRequest != null) {
          LOG.warn("Close a closed block {}.", mRequest.getBlockId());
        }
        return null;
      }

      @Override
      public void exceptionCaught(Throwable e) {
        mResponseObserver.onError(GrpcExceptionUtils.fromThrowable(e));
        mLockId = BlockLockManager.INVALID_LOCK_ID;
      }
    }, "CloseBlock", false, true, mResponseObserver, "Session=%d, Request=%s",
        mSessionId, mRequest);
  }
181918
@FFDCIgnore({WebApplicationException.class, IOException.class, NoSuitableMessageBodyWriterException.class})
    @Override
    public CompletionStage<?> send(OutboundSseEvent event) {
        final CompletableFuture<?> future = new CompletableFuture<>();

        if (!closed) {
            if (writer != null) {
                ByteArrayOutputStream os = null;
                try {
                    os = new ByteArrayOutputStream();
                    writer.writeTo(event, event.getClass(), null, new Annotation [] {}, event.getMediaType(), null, os);

                    String eventContents = os.toString();
                    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                        Tr.debug(tc, "send - sending " + eventContents);
                    }

                    if (!response.isCommitted()) {
                        response.setHeader("Content-Type", MediaType.SERVER_SENT_EVENTS);
                        response.flushBuffer();
                    }

                    //TODO: this seems like a bug, but most SSE clients seem to expect a named event
                    //      so for now, we will provide one if one is not provided by the user
                    if (event.getName() == null) {
                        response.getOutputStream().print("    UnnamedEvent\n");
                    }
                    response.getOutputStream().println(eventContents);
                    response.getOutputStream().flush();

                    return CompletableFuture.completedFuture(eventContents);
                } catch (NoSuitableMessageBodyWriterException ex) {
                    handleException(ex, future, event);
                    throw new IllegalArgumentException("No suitable message body writer for OutboundSseEvent created with data " + event.getData() + " and mediaType " + event.getMediaType() + ". The data contained within the OutboundSseEvent must match the mediaType."); 
                } catch (WebApplicationException ex) {
                    handleException(ex, future, event);
                } catch (IOException ex) {
                    handleException(ex, future, event);
                } finally {
                    if (os != null) {
                        try {
                            os.close();
                        } catch (IOException ex) {
                            //ignore
                        }
                    }
                }
            } else {  //no writer
                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                    Tr.debug(tc, "No MessageBodyWriter - returning null for event:  " + event);
                }
                future.complete(null);
            }  
        } else {
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                Tr.debug(tc, "SseEventSink is closed - failed sending event:  " + event);
            }
            throw new IllegalStateException("SseEventSink is closed.");  
        }

        return future;
    }
243282
synchronized void requestCompleted(ClientWorldModelInterface worldModel,
      AbstractRequestMessage message) {
    Long ticket = Long.valueOf(message.getTicketNumber());
    log.debug("Request {} completed.", ticket);

    Response resp = this.outstandingSnapshots.remove(ticket);
    // Snapshot request
    if (resp != null) {
      log.debug("Retrieved {} for Tix#{}", resp, ticket);
      WorldState ws = this.outstandingStates.get(ticket);
      if (ws == null) {
        log.error("Unknown ticket number {} for request.", ticket, message);
        return;
      }
      resp.setState(ws);
      return;
    }

    StepResponse sResp = this.outstandingSteps.remove(ticket);
    // Range/Streaming request
    if (sResp != null) {
      log.debug("Retrieved {} for Tix#{}", resp, ticket);
      sResp.setComplete();
      return;
    }

    log.error("Couldn't find response for ticket {}.", ticket);
  }
1117464
private void handleInCompletedState(final SipPacket msg) throws SipPacketParseException {
        if (msg.isRequest()) {
            // TODO:
        } else {
            if (msg.isBye()) {
                transition(CallState.COMPLETED, msg);
            }
        }
    }
716072
function onPeerClose(peer)
    {
        var id = peer.id;
        scope.removePeer(peer);
        sendPeerClose(id);

        scope.dispatchEvent("peerDisconnected",id);
    }
464990
private void waitForAsyncSendCompletion() {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "waitForAsyncSendCompletion");

        while (!(_asyncSendQueue.isEmpty() && (currentAsyncSendObject == null))) {
            try {
                synchronized (_asyncSendQueue) {
                    // after twenty second.. again check is made if the async send queue is empty.
                    //the sleep time is precautinary measure.. to avoid a futuristic scenario in which notification would not come.
                    _asyncSendQueue.wait(1000 * 20);
                }
            } catch (InterruptedException e) {
                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                    SibTr.debug(tc, "waitForAsyncSendCompletion got interrupted", e);
            }
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(this, tc, "waitForAsyncSendCompletion");
    }
1365877
private void onParticipantAdded(ParticipantAddedEvent event) {
        handler.post(() -> listener.onParticipantAdded(event));
        log("Event published " + event.toString());
    }
658306
@Override
    public void afterCompletion(TransactionCommon transaction, boolean committed)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "afterCompletion", new Object[] { transaction, Boolean.valueOf(committed) });

        synchronized (this)
        {
            // If the connection is already closed this transaction
            // will have been removed.
            if (!_closed)
            {
                synchronized (_ownedTransactions)
                {
                    // Remove the transaction from the connections list
                    _ownedTransactions.remove(transaction);
                }
            }
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "afterCompletion");
    }
695876
public void onS3UploadComplete(S3UploadEvent uploadEvent) {
        if (VERBOSE) Log.i(TAG, "Upload completed for " + uploadEvent.getDestinationUrl());
        if (uploadEvent.getDestinationUrl().contains(".m3u8")) {
            onManifestUploaded(uploadEvent);
        } else if (uploadEvent.getDestinationUrl().contains(".ts")) {
            onSegmentUploaded(uploadEvent);
        } else if (uploadEvent.getDestinationUrl().contains(".jpg")) {
            onThumbnailUploaded(uploadEvent);
        }
    }
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);
                }
            }
        });
    }
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;
  }
754482
@Override
    public void afterCompletion(TransactionCommon transaction, boolean committed)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "afterCompletion", new Object[] { transaction, Boolean.valueOf(committed) });

        // Reset the current transaction now that this one has completed
        synchronized (orderLock)
        {
            // If asynch and a rollback occurred, unlock all messages
            // in the LME
            try
            {
                if (!committed && currentLME != null)
                    currentLME.unlockAll();
            } catch (SIResourceException e)
            {
                // FFDC
                FFDCFilter.processException(
                                            e,
                                            "com.ibm.ws.sib.processor.impl.ConsumerDispatcher.afterCompletion",
                                            "1:3673:1.280.5.25",
                                            this);

                SibTr.exception(tc, e);
                SibTr.error(tc, "ORDERED_MESSAGING_ERROR_CWSIP0117",
                            new Object[] {
                                          _baseDestHandler.getName(),
                                          _messageProcessor.getMessagingEngineName() });

            } catch (SISessionDroppedException e)
            {
                // FFDC
                FFDCFilter.processException(
                                            e,
                                            "com.ibm.ws.sib.processor.impl.ConsumerDispatcher.afterCompletion",
                                            "1:3689:1.280.5.25",
                                            this);

                SibTr.exception(tc, e);
                SibTr.error(tc, "ORDERED_MESSAGING_ERROR_CWSIP0117",
                            new Object[] {
                                          _baseDestHandler.getName(),
                                          _messageProcessor.getMessagingEngineName() });

            } catch (SIMPMessageNotLockedException e)
            {
                // No FFDC code needed
                // This exception has occurred beause someone has deleted the
                // message(s). Ignore this exception as it is unlocked anyway
            }

            currentTran = null;
            currentLME = null;
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "afterCompletion");
    }
1363867
private CompletionStage<Void> emitMembershipUpdateEvent() {
        final IRI membershipResource = parent.getMembershipResource().map(MutatingLdpHandler::removeHashFragment)
            .orElse(null);
        if (membershipResource != null) {
            return allOf(getServices().getResourceService().touch(membershipResource).toCompletableFuture(),
                getServices().getResourceService().get(membershipResource).thenAccept(res -> {
                    if (res.getIdentifier() != null) {
                        getServices().getEventService().emit(new SimpleEvent(getUrl(res.getIdentifier()),
                                    getSession().getAgent(), asList(PROV.Activity, AS.Update),
                                    asList(res.getInteractionModel())));
                    }
                }).toCompletableFuture());
        }
        return completedFuture(null);
    }
782570
private void firePieceCompleted(Piece piece) throws IOException {
    for (PeerActivityListener listener : this.listeners) {
      listener.handlePieceCompleted(this, piece);
    }
  }
833478
synchronized void processFinished() {
        final InternalState required = this.requiredState;
        final InternalState state = this.internalState;
        // If the server was not stopped
        if(required == InternalState.STOPPED && state == InternalState.PROCESS_STOPPING) {
            finishTransition(InternalState.PROCESS_STOPPING, InternalState.PROCESS_STOPPED);
        } else {
            this.requiredState = InternalState.STOPPED;
            if ( !(internalSetState(getTransitionTask(InternalState.PROCESS_STOPPING), internalState, InternalState.PROCESS_STOPPING)
                    && internalSetState(getTransitionTask(InternalState.PROCESS_REMOVING), internalState, InternalState.PROCESS_REMOVING)
                    && internalSetState(getTransitionTask(InternalState.STOPPED), internalState, InternalState.STOPPED)) ){
                this.requiredState = InternalState.FAILED;
                internalSetState(null, internalState, InternalState.PROCESS_STOPPED);
            }
        }
    }