import com.twitter.util.FutureEventListener; //导入依赖的package包/类
public void truncateLogBeforeDLSN(final DLSN dlsn) {
    if (dlsn.getLogSegmentSequenceNo() <= truncatedDlsn.getLogSegmentSequenceNo()) {
        return;
    }
    this.logWriter.truncate(dlsn).addEventListener(new FutureEventListener<Boolean>() {

        @Override
        public void onFailure(Throwable t) {
            logger.error("errors while truncate log after DLSN [{}]", t, dlsn);
        }

        @Override
        public void onSuccess(Boolean isSuccess) {
            if (isSuccess) {
                truncatedDlsn = dlsn;
                logger.info("truncate log before [{}]", dlsn);
            }
        }
    });
}
 

import com.twitter.util.FutureEventListener; //导入依赖的package包/类
public void truncateLogBeforeDLSN(final DLSN dlsn) {
    if (dlsn.getLogSegmentSequenceNo() <= truncatedDlsn.getLogSegmentSequenceNo()) {
        logger.info("previous truncate dlsn is [{}], current dlsn is [{}], segment no not change not call truncate", truncatedDlsn, dlsn);
        return;
    }
    if (this.logWriter == null) {
        logger.error("log writer is closed, maybe not primary any more, skip truncate");
        return;
    }
    this.logWriter.truncate(dlsn).addEventListener(new FutureEventListener<Boolean>() {

        @Override
        public void onFailure(Throwable t) {
            logger.error("errors while truncate log after DLSN [{}]", t, dlsn);
        }

        @Override
        public void onSuccess(Boolean isSuccess) {
            if (isSuccess) {
                truncatedDlsn = dlsn;
                logger.info("truncate log before [{}] successfully", dlsn);
            }
        }
    });
}
 

import com.twitter.util.FutureEventListener; //导入依赖的package包/类
/**
 * Create a client to proxy <code>address</code>.
 *
 * @param address
 *          proxy address
 * @return proxy client
 */
public ProxyClient createClient(final SocketAddress address) {
    final ProxyClient sc = clientBuilder.build(address);
    ProxyClient oldSC = address2Services.putIfAbsent(address, sc);
    if (null != oldSC) {
        sc.close();
        return oldSC;
    } else {
        final Stopwatch stopwatch = Stopwatch.createStarted();
        FutureEventListener<ServerInfo> listener = new FutureEventListener<ServerInfo>() {
            @Override
            public void onSuccess(ServerInfo serverInfo) {
                notifyHandshakeSuccess(address, sc, serverInfo, true, stopwatch);
            }
            @Override
            public void onFailure(Throwable cause) {
                notifyHandshakeFailure(address, sc, cause, stopwatch);
            }
        };
        // send a ping messaging after creating connections.
        handshake(address, sc, listener, true, true);
        return sc;
    }
}
 

import com.twitter.util.FutureEventListener; //导入依赖的package包/类
/**
 * Handshake with a given proxy
 *
 * @param address
 *          proxy address
 * @param sc
 *          proxy client
 * @param listener
 *          listener on handshake result
 */
private void handshake(SocketAddress address,
                       ProxyClient sc,
                       FutureEventListener<ServerInfo> listener,
                       boolean logging,
                       boolean getOwnerships) {
    if (clientConfig.getHandshakeWithClientInfo()) {
        ClientInfo clientInfo = new ClientInfo();
        clientInfo.setGetOwnerships(getOwnerships);
        clientInfo.setStreamNameRegex(clientConfig.getStreamNameRegex());
        if (logging) {
            logger.info("Handshaking with {} : {}", address, clientInfo);
        }
        sc.getService().handshakeWithClientInfo(clientInfo)
                .addEventListener(listener);
    } else {
        if (logging) {
            logger.info("Handshaking with {}", address);
        }
        sc.getService().handshake().addEventListener(listener);
    }
}
 

import com.twitter.util.FutureEventListener; //导入依赖的package包/类
private void fetchDefaultAccessControlEntry(final Promise<ZKAccessControl> promise) {
    ZKAccessControl.read(zkc, zkRootPath, this)
        .addEventListener(new FutureEventListener<ZKAccessControl>() {
            @Override
            public void onSuccess(ZKAccessControl accessControl) {
                logger.info("Default Access Control will be changed from {} to {}",
                            ZKAccessControlManager.this.defaultAccessControl,
                            accessControl);
                ZKAccessControlManager.this.defaultAccessControl = accessControl;
                promise.setValue(accessControl);
            }

            @Override
            public void onFailure(Throwable cause) {
                if (cause instanceof KeeperException.NoNodeException) {
                    logger.info("Default Access Control is missing, creating one for {} ...", zkRootPath);
                    createDefaultAccessControlEntryIfNeeded(promise);
                } else {
                    promise.setException(cause);
                }
            }
        });
}
 

import com.twitter.util.FutureEventListener; //导入依赖的package包/类
private void refetchAccessControlEntries(final int delayMs) {
    if (closed) {
        return;
    }
    scheduledExecutorService.schedule(new Runnable() {
        @Override
        public void run() {
            fetchAccessControlEntries().addEventListener(new FutureEventListener<Void>() {
                @Override
                public void onSuccess(Void value) {
                    // no-op
                }
                @Override
                public void onFailure(Throwable cause) {
                    logger.warn("Encountered an error on refetching access control entries, retrying in {} ms : ",
                                ZK_RETRY_BACKOFF_MS, cause);
                    refetchAccessControlEntries(ZK_RETRY_BACKOFF_MS);
                }
            });
        }
    }, delayMs, TimeUnit.MILLISECONDS);
}
 

import com.twitter.util.FutureEventListener; //导入依赖的package包/类
private void markAsAllocated(final LedgerHandle lh) {
    byte[] data = DLUtils.ledgerId2Bytes(lh.getId());
    Utils.zkSetData(zkc, allocatePath, data, getVersion())
        .addEventListener(new FutureEventListener<ZkVersion>() {
            @Override
            public void onSuccess(ZkVersion version) {
                // we only issue deleting ledger left from previous allocation when we could allocate first ledger
                // as zookeeper version could prevent us doing stupid things.
                deleteLedgerLeftFromPreviousAllocationIfNecessary();
                setVersion(version);
                setPhase(Phase.ALLOCATED);
                // complete the allocation after it is marked as allocated
                completeAllocation(lh);
            }

            @Override
            public void onFailure(Throwable cause) {
                setPhase(Phase.ERROR);
                deleteLedger(lh.getId());
                LOG.error("Fail mark ledger {} as allocated under {} : ",
                        new Object[] { lh.getId(), allocatePath, cause });
                // fail the allocation since failed to mark it as allocated
                failAllocation(cause);
            }
        });
}
 

import com.twitter.util.FutureEventListener; //导入依赖的package包/类
void waitForAcquire(final LockWaiter waiter,
                    final Promise<ZKDistributedLock> acquirePromise) {
    waiter.getAcquireFuture().addEventListener(OrderedFutureEventListener.of(
            new FutureEventListener<Boolean>() {
                @Override
                public void onSuccess(Boolean acquired) {
                    LOG.info("{} acquired lock {}", waiter, lockPath);
                    if (acquired) {
                        FutureUtils.setValue(acquirePromise, ZKDistributedLock.this);
                    } else {
                        FutureUtils.setException(acquirePromise,
                                new OwnershipAcquireFailedException(lockPath, waiter.getCurrentOwner()));
                    }
                }

                @Override
                public void onFailure(Throwable cause) {
                    FutureUtils.setException(acquirePromise, cause);
                }
            }, lockStateExecutor, lockPath));
}
 

import com.twitter.util.FutureEventListener; //导入依赖的package包/类
void closeWaiter(final LockWaiter waiter,
                 final Promise<Void> closePromise) {
    if (null == waiter) {
        interruptTryLock(tryLockFuture, closePromise);
    } else {
        waiter.getAcquireFuture().addEventListener(OrderedFutureEventListener.of(
                new FutureEventListener<Boolean>() {
                    @Override
                    public void onSuccess(Boolean value) {
                        unlockInternalLock(closePromise);
                    }
                    @Override
                    public void onFailure(Throwable cause) {
                        unlockInternalLock(closePromise);
                    }
                }, lockStateExecutor, lockPath));
        FutureUtils.cancel(waiter.getAcquireFuture());
    }
}
 

import com.twitter.util.FutureEventListener; //导入依赖的package包/类
void interruptTryLock(final Future<LockWaiter> tryLockFuture,
                      final Promise<Void> closePromise) {
    if (null == tryLockFuture) {
        unlockInternalLock(closePromise);
    } else {
        tryLockFuture.addEventListener(OrderedFutureEventListener.of(
                new FutureEventListener<LockWaiter>() {
                    @Override
                    public void onSuccess(LockWaiter waiter) {
                        closeWaiter(waiter, closePromise);
                    }
                    @Override
                    public void onFailure(Throwable cause) {
                        unlockInternalLock(closePromise);
                    }
                }, lockStateExecutor, lockPath));
        FutureUtils.cancel(tryLockFuture);
    }
}
 

import com.twitter.util.FutureEventListener; //导入依赖的package包/类
/**
 * Ignore exception from the <i>future</i> and log <i>errorMsg</i> on exceptions
 *
 * @param future the original future
 * @param errorMsg the error message to log on exceptions
 * @return a transformed future ignores exceptions
 */
public static <T> Promise<Void> ignore(Future<T> future, final String errorMsg) {
    final Promise<Void> promise = new Promise<Void>();
    future.addEventListener(new FutureEventListener<T>() {
        @Override
        public void onSuccess(T value) {
            setValue(promise, null);
        }

        @Override
        public void onFailure(Throwable cause) {
            if (null != errorMsg) {
                logger.error(errorMsg, cause);
            }
            setValue(promise, null);
        }
    });
    return promise;
}
 

import com.twitter.util.FutureEventListener; //导入依赖的package包/类
private <T> Future<T> postStateCheck(Future<T> future) {
    final Promise<T> postCheckedPromise = new Promise<T>();
    future.addEventListener(new FutureEventListener<T>() {
        @Override
        public void onSuccess(T value) {
            if (duplicatedLogFound.get()) {
                postCheckedPromise.setException(new UnexpectedException("Duplicate log found under " + namespace));
            } else {
                postCheckedPromise.setValue(value);
            }
        }

        @Override
        public void onFailure(Throwable cause) {
            postCheckedPromise.setException(cause);
        }
    });
    return postCheckedPromise;
}
 

import com.twitter.util.FutureEventListener; //导入依赖的package包/类
void doCreateLog(final String logName, final Promise<URI> createPromise) {
    getLogLocation(logName).addEventListener(new FutureEventListener<Optional<URI>>() {
        @Override
        public void onSuccess(Optional<URI> uriOptional) {
            if (uriOptional.isPresent()) {
                createPromise.setException(new LogExistsException("Log " + logName + " already exists in " + uriOptional.get()));
            } else {
                getCachedSubNamespacesAndCreateLog(logName, createPromise);
            }
        }

        @Override
        public void onFailure(Throwable cause) {
            createPromise.setException(cause);
        }
    });
}
 

import com.twitter.util.FutureEventListener; //导入依赖的package包/类
protected Future<List<LogSegmentMetadata>> asyncGetLedgerList(final boolean forceFetch,
                                                                    final boolean fetchFullList,
                                                                    final Comparator<LogSegmentMetadata> comparator,
                                                                    final boolean throwOnEmpty) {
    final Promise<List<LogSegmentMetadata>> promise = new Promise<List<LogSegmentMetadata>>();
    final Stopwatch stopwatch = Stopwatch.createStarted();
    final OpStatsLogger statsLogger = fetchFullList ? getFullListStat : getFilteredListStat;
    asyncDoGetLedgerList(forceFetch, fetchFullList, comparator, throwOnEmpty)
            .addEventListener(new FutureEventListener<List<LogSegmentMetadata>>() {
                @Override
                public void onSuccess(List<LogSegmentMetadata> value) {
                    statsLogger.registerSuccessfulEvent(stopwatch.stop().elapsed(TimeUnit.MICROSECONDS));
                    promise.setValue(value);
                }

                @Override
                public void onFailure(Throwable cause) {
                    statsLogger.registerFailedEvent(stopwatch.stop().elapsed(TimeUnit.MICROSECONDS));
                    promise.setException(cause);
                }
            });
    return promise;
}
 

import com.twitter.util.FutureEventListener; //导入依赖的package包/类
protected Future<List<LogSegmentMetadata>> asyncForceGetLedgerList(final Comparator<LogSegmentMetadata> comparator,
                                                                   final LogSegmentFilter segmentFilter,
                                                                   final boolean throwOnEmpty) {
    final Promise<List<LogSegmentMetadata>> promise = new Promise<List<LogSegmentMetadata>>();
    final Stopwatch stopwatch = Stopwatch.createStarted();
    asyncGetLedgerListWithRetries(comparator, segmentFilter, null)
        .addEventListener(new FutureEventListener<List<LogSegmentMetadata>>() {

            @Override
            public void onSuccess(List<LogSegmentMetadata> ledgers) {
                forceGetListStat.registerSuccessfulEvent(stopwatch.stop().elapsed(TimeUnit.MICROSECONDS));
                if (ledgers.isEmpty() && throwOnEmpty) {
                    promise.setException(new LogEmptyException("Log " + getFullyQualifiedName() + " is empty"));
                } else {
                    promise.setValue(ledgers);
                }
            }

            @Override
            public void onFailure(Throwable cause) {
                forceGetListStat.registerFailedEvent(stopwatch.stop().elapsed(TimeUnit.MICROSECONDS));
                promise.setException(cause);
            }
        });
    return promise;
}
 

import com.twitter.util.FutureEventListener; //导入依赖的package包/类
private Future<BKLogSegmentWriter> completeOldSegmentAndCacheNewLogSegmentWriter(
        BKLogSegmentWriter oldSegmentWriter,
        final BKLogSegmentWriter newSegmentWriter) {
    final Promise<BKLogSegmentWriter> completePromise = new Promise<BKLogSegmentWriter>();
    // complete the old log segment
    writeHandler.completeAndCloseLogSegment(oldSegmentWriter)
            .addEventListener(new FutureEventListener<LogSegmentMetadata>() {

                @Override
                public void onSuccess(LogSegmentMetadata value) {
                    cacheLogWriter(newSegmentWriter);
                    removeAllocatedLogWriter();
                    FutureUtils.setValue(completePromise, newSegmentWriter);
                }

                @Override
                public void onFailure(Throwable cause) {
                    FutureUtils.setException(completePromise, cause);
                }
            });
    return completePromise;
}
 

import com.twitter.util.FutureEventListener; //导入依赖的package包/类
private static void readLoop(final DistributedLogManager dlm,
                             final DLSN dlsn,
                             final CountDownLatch keepAliveLatch) {
    System.out.println("Wait for records from " + dlm.getStreamName() + " starting from " + dlsn);
    dlm.openAsyncLogReader(dlsn).addEventListener(new FutureEventListener<AsyncLogReader>() {
        @Override
        public void onFailure(Throwable cause) {
            System.err.println("Encountered error on reading records from stream " + dlm.getStreamName());
            cause.printStackTrace(System.err);
            keepAliveLatch.countDown();
        }

        @Override
        public void onSuccess(AsyncLogReader reader) {
            System.out.println("Open reader to read records from stream " + reader.getStreamName());
            readLoop(reader, keepAliveLatch);
        }
    });
}
 

import com.twitter.util.FutureEventListener; //导入依赖的package包/类
DLFutureRecordMetadata(final String topic,
                       com.twitter.util.Future<DLSN> dlsnFuture,
                       final Callback callback) {
    this.topic = topic;
    this.dlsnFuture = dlsnFuture;
    this.callback = callback;

    this.dlsnFuture.addEventListener(new FutureEventListener<DLSN>() {
        @Override
        public void onFailure(Throwable cause) {
            callback.onCompletion(null, new IOException(cause));
        }

        @Override
        public void onSuccess(DLSN value) {
            callback.onCompletion(new RecordMetadata(new TopicPartition(topic, 0), -1L, -1L), null);
        }
    });
}
 

import com.twitter.util.FutureEventListener; //导入依赖的package包/类
@Override
public void run() {
    final DLSN dlsnToTruncate = prevDLSN;
    if (null == dlsnToTruncate) {
        return;
    }
    final Stopwatch stopwatch = Stopwatch.createStarted();
    dlc.truncate(streamName, dlsnToTruncate).addEventListener(
            new FutureEventListener<Boolean>() {
                @Override
                public void onSuccess(Boolean value) {
                    truncationStat.registerSuccessfulEvent(stopwatch.stop().elapsed(TimeUnit.MILLISECONDS));
                }

                @Override
                public void onFailure(Throwable cause) {
                    truncationStat.registerFailedEvent(stopwatch.stop().elapsed(TimeUnit.MILLISECONDS));
                    LOG.error("Failed to truncate stream {} to {} : ",
                            new Object[]{streamName, dlsnToTruncate, cause});
                }
            });
}
 

import com.twitter.util.FutureEventListener; //导入依赖的package包/类
/**
 * Asynchronously runs video generation job in thread pool
 */
private void generateVideo(Map<Object, Object> jobInfo) {
    Future<Object> videoGenResultF = futurePool.apply(new VideoGenerator(jobInfo));
    
    videoGenResultF.addEventListener(new FutureEventListener<Object>() {

        public void onFailure(Throwable e) {
            e.printStackTrace();
        }

        public void onSuccess(Object obj) {
            notifySuccess(obj);
        }
        
    });
}