import com.google.common.util.concurrent.JdkFutureAdapters; //导入依赖的package包/类
private void createIdPool() {
    if (checkPoolExists()) {
        return;
    }
    CreateIdPoolInput createPool = new CreateIdPoolInputBuilder()
            .setPoolName(CountersServiceUtils.COUNTERS_PULL_NAME).setLow(CountersServiceUtils.COUNTERS_PULL_START)
            .setHigh(CountersServiceUtils.COUNTERS_PULL_START + CountersServiceUtils.COUNTERS_PULL_END).build();
    Future<RpcResult<Void>> result = idManagerService.createIdPool(createPool);
    Futures.addCallback(JdkFutureAdapters.listenInPoolThread(result), new FutureCallback<RpcResult<Void>>() {

        @Override
        public void onFailure(Throwable error) {
            LOG.error("Failed to create idPool for Aliveness Monitor Service", error);
        }

        @Override
        public void onSuccess(@Nonnull RpcResult<Void> rpcResult) {
            if (rpcResult.isSuccessful()) {
                LOG.debug("Created IdPool for tap");
            } else {
                LOG.error("RPC to create Idpool failed {}", rpcResult.getErrors());
            }
        }
    }, MoreExecutors.directExecutor());
}
 

import com.google.common.util.concurrent.JdkFutureAdapters; //导入依赖的package包/类
/**
 * Removes the corresponding static routes from the specified VPN. These static routes point to an
 * InterVpnLink endpoint and the specified VPN must be the other end of the InterVpnLink.
 *
 * @param vpnName the VPN identifier
 * @param interVpnLinkRoutes The list of static routes
 * @param nexthopsXinterVpnLinks A Map with the correspondence nextHop-InterVpnLink
 */
public void removeInterVpnRoutes(Uuid vpnName, List<Routes> interVpnLinkRoutes,
                                 HashMap<String, InterVpnLink> nexthopsXinterVpnLinks) {
    for (Routes route : interVpnLinkRoutes) {
        String nexthop = String.valueOf(route.getNexthop().getValue());
        String destination = String.valueOf(route.getDestination().getValue());
        InterVpnLink interVpnLink = nexthopsXinterVpnLinks.get(nexthop);
        if (isNexthopTheOtherVpnLinkEndpoint(nexthop, vpnName.getValue(), interVpnLink)) {
            RemoveStaticRouteInput rpcInput =
                    new RemoveStaticRouteInputBuilder().setDestination(destination).setNexthop(nexthop)
                            .setVpnInstanceName(vpnName.getValue())
                            .build();

            ListenableFutures.addErrorLogging(JdkFutureAdapters.listenInPoolThread(
                    vpnRpcService.removeStaticRoute(rpcInput)), LOG, "Remove VPN routes");
        } else {
            // Any other case is a fault.
            LOG.warn("route with destination {} and nexthop {} does not apply to any InterVpnLink",
                    String.valueOf(route.getDestination().getValue()), nexthop);
            continue;
        }
    }
}
 

import com.google.common.util.concurrent.JdkFutureAdapters; //导入依赖的package包/类
@SuppressWarnings("checkstyle:IllegalCatch")
private void sendArpRequest(IpAddress srcIpAddress, IpAddress dstIpAddress, MacAddress srcMacAddress,
        String interfaceName) {
    if (srcIpAddress == null || dstIpAddress == null) {
        LOG.trace("Skip sending ARP to external GW srcIp {} dstIp {}", srcIpAddress, dstIpAddress);
        return;
    }

    PhysAddress srcMacPhysAddress = new PhysAddress(srcMacAddress.getValue());
    try {
        InterfaceAddress interfaceAddress = new InterfaceAddressBuilder().setInterface(interfaceName)
                .setIpAddress(srcIpAddress).setMacaddress(srcMacPhysAddress).build();

        SendArpRequestInput sendArpRequestInput = new SendArpRequestInputBuilder().setIpaddress(dstIpAddress)
                .setInterfaceAddress(Collections.singletonList(interfaceAddress)).build();

        ListenableFutures.addErrorLogging(JdkFutureAdapters.listenInPoolThread(
                arpUtilService.sendArpRequest(sendArpRequestInput)), LOG, "Send ARP request");
    } catch (Exception e) {
        LOG.error("Failed to send ARP request to external GW {} from interface {}",
                dstIpAddress.getIpv4Address().getValue(), interfaceName, e);
    }
}
 

import com.google.common.util.concurrent.JdkFutureAdapters; //导入依赖的package包/类
private void sendPacketOut(byte[] payload, NodeConnectorRef egress) {
	if (egress == null) {
		LOG.info("Egress is null");
		return;
	}

	InstanceIdentifier<Node> egressNodePath = getNodePath(egress.getValue());

	TransmitPacketInput input = new TransmitPacketInputBuilder() //
			.setPayload(payload) //
			.setNode(new NodeRef(egressNodePath)) //
			.setEgress(egress) //
			.build();

	Future<RpcResult<Void>> future = packetService.transmitPacket(input);
	JdkFutureAdapters.listenInPoolThread(future);
}
 

import com.google.common.util.concurrent.JdkFutureAdapters; //导入依赖的package包/类
public ListenableFuture<RpcResult<Void>> sendArpResponse(ArpMessageAddress senderAddress,
		ArpMessageAddress receiverAddress, InstanceIdentifier<NodeConnector> egressNc, Header8021q vlan) {
	checkNotNull(senderAddress);
	checkNotNull(receiverAddress);
	checkNotNull(egressNc);
	final Ethernet arpFrame = createArpFrame(senderAddress, receiverAddress, vlan);
	byte[] arpFrameAsBytes;
	try {
		arpFrameAsBytes = arpFrame.serialize();
	} catch (PacketException e) {
		LOG.warn("Serializition of ARP packet is not successful.", e);
		if (LOG.isDebugEnabled()) {
			LOG.debug("ARP packet: {}", ArpUtils.getArpFrameToStringFormat(arpFrame));
		}
		return Futures.immediateFailedFuture(e);
	}
	// Generate packet with destination switch and port
	LOG.debug("Egress for ARP packetOut: " + new NodeConnectorRef(egressNc).toString());
	TransmitPacketInput packet = new TransmitPacketInputBuilder().setEgress(new NodeConnectorRef(egressNc))
			.setNode(new NodeRef(egressNc.firstIdentifierOf(Node.class))).setPayload(arpFrameAsBytes).build();
	if (LOG.isTraceEnabled()) {
		LOG.trace("Sending ARP RESPONSE \n{}", ArpUtils.getArpFrameToStringFormat(arpFrame));
	}
	Future<RpcResult<Void>> futureTransmitPacketResult = packetProcessingService.transmitPacket(packet);
	return JdkFutureAdapters.listenInPoolThread(futureTransmitPacketResult);
}
 

import com.google.common.util.concurrent.JdkFutureAdapters; //导入依赖的package包/类
@PostConstruct
public void recreateIndex() throws Exception {
	log.debug("SearchConfiguration - creating full-text index");
	
	FullTextEntityManager fullTextEntityManager = Search.getFullTextEntityManager(entityManager);
	
	Future<?> indexResult = fullTextEntityManager.createIndexer().start();
	
	Futures.addCallback(JdkFutureAdapters.listenInPoolThread(indexResult), new FutureCallback<Object>() {
		@Override
		public void onSuccess(Object result) {
			log.debug("SearchConfiguration - finished creating full-text index");
		}

		@Override
		public void onFailure(Throwable t) {
			log.error("SearchConfiguration - Failed creating full-text index", t);
		}
	});
}
 

import com.google.common.util.concurrent.JdkFutureAdapters; //导入依赖的package包/类
public <E extends DataObject> void sendResponseToCore(Future<RpcResult<E>> switchReply,
        final ZeroMQBaseConnector coreConnector, final short ofVersion, final long xId, final long datapathId,
        final int moduleId) {

    Futures.addCallback(JdkFutureAdapters.listenInPoolThread(switchReply), new FutureCallback<RpcResult<E>>() {
        @Override
        public void onSuccess(RpcResult<E> rpcReply) {
            if (rpcReply.isSuccessful()) {
                E result = rpcReply.getResult();

                sendOpenFlowMessageToCore(coreConnector, result, ofVersion, xId, datapathId, moduleId);
            } else {
                for (RpcError rpcError : rpcReply.getErrors()) {
                    LOG.info("SHIM RELAY: error in communication with switch: {}", rpcError.getMessage());
                }
            }
        }

        @Override
        public void onFailure(Throwable t) {
            LOG.info("SHIM RELAY: failure on communication with switch");
        }
    });
}
 

import com.google.common.util.concurrent.JdkFutureAdapters; //导入依赖的package包/类
/**
 * Converts the given {@link Future} object to a {@link RedFuture}
 * see {@link JdkFutureAdapters#listenInPoolThread(Future)} for detailed implications
 * @param future future to convert
 * @param <T>    type of the future value
 * @return a RedFuture instance tracking the given {@link Future}
 */
static <T> RedFutureOf<T> convert(Future<T> future) {
    if (future instanceof RedFutureOf) {
        return (RedFutureOf<T>) future;
    }
    if (future instanceof ListenableFuture) {
        OpenRedFutureOf<T> result = futureOf();
        result.follow((ListenableFuture<T>) future);
        return result;
    }
    return convert(JdkFutureAdapters.listenInPoolThread(future));
}
 

import com.google.common.util.concurrent.JdkFutureAdapters; //导入依赖的package包/类
@Override
public Future<RpcResult<Void>> makeBreakfast(EggsType eggsType, Class<? extends ToastType> toastType,
        int toastDoneness) {
    // Call makeToast and use JdkFutureAdapters to convert the Future to a ListenableFuture, The
    // OpendaylightToaster impl already returns a ListenableFuture so the conversion is actually a no-op.

    ListenableFuture<RpcResult<Void>> makeToastFuture = JdkFutureAdapters
            .listenInPoolThread(makeToast(toastType, toastDoneness), executor);

    ListenableFuture<RpcResult<Void>> makeEggsFuture = makeEggs(eggsType);

    // Combine the 2 ListenableFutures into 1 containing a list RpcResults.

    ListenableFuture<List<RpcResult<Void>>> combinedFutures = Futures
            .allAsList(ImmutableList.of(makeToastFuture, makeEggsFuture));

    // Then transform the RpcResults into 1.

    return Futures.transformAsync(combinedFutures, results -> {
        boolean atLeastOneSucceeded = false;
        Builder<RpcError> errorList = ImmutableList.builder();
        for (RpcResult<Void> result : results) {
            if (result.isSuccessful()) {
                atLeastOneSucceeded = true;
            }

            if (result.getErrors() != null) {
                errorList.addAll(result.getErrors());
            }
        }

        return Futures.immediateFuture(RpcResultBuilder.<Void>status(atLeastOneSucceeded)
                .withRpcErrors(errorList.build()).build());
    });
}
 

import com.google.common.util.concurrent.JdkFutureAdapters; //导入依赖的package包/类
@SafeVarargs
private final <V> ListenableFuture<List<V>> allAsList(Future<? extends V>... futures) {
  ImmutableList.Builder<ListenableFuture<? extends V>> listBuilder = ImmutableList.builder();
  for (Future<? extends V> future : futures) {
    listBuilder.add(JdkFutureAdapters.listenInPoolThread(future));
  }
  return Futures.allAsList(listBuilder.build());
}
 

import com.google.common.util.concurrent.JdkFutureAdapters; //导入依赖的package包/类
@Override
public ListenableFuture<String> asyncGet(final String url) {
  return JdkFutureAdapters.listenInPoolThread(executor.submit(new Callable<String>() {
    @Override
    public String call() throws RobotConnectionException {
      return get(url);
    }
  }));
}
 

import com.google.common.util.concurrent.JdkFutureAdapters; //导入依赖的package包/类
@Override
public ListenableFuture<String> asyncPostJson(final String url, final String body) {
  return JdkFutureAdapters.listenInPoolThread(executor.submit(new Callable<String>() {
    @Override
    public String call() throws RobotConnectionException {
      return postJson(url, body);
    }
  }));
}
 

import com.google.common.util.concurrent.JdkFutureAdapters; //导入依赖的package包/类
/**
 * Iterates through all {@link TargetFinder}s, returning a {@link Future} representing the first
 * non-null result.
 */
static ListenableFuture<TargetInfo> findTargetInfoFuture(Project project, Label label) {
  List<ListenableFuture<TargetInfo>> futures = new ArrayList<>();
  for (TargetFinder finder : EP_NAME.getExtensions()) {
    Future<TargetInfo> future = finder.findTarget(project, label);
    if (future.isDone() && futures.isEmpty()) {
      TargetInfo target = FuturesUtil.getIgnoringErrors(future);
      if (target != null) {
        return Futures.immediateFuture(target);
      }
    } else {
      // we can't return ListenableFuture directly, because implementations are using different
      // versions of that class...
      futures.add(JdkFutureAdapters.listenInPoolThread(future));
    }
  }
  if (futures.isEmpty()) {
    return Futures.immediateFuture(null);
  }
  return Futures.transform(
      Futures.allAsList(futures),
      (Function<List<TargetInfo>, TargetInfo>)
          list ->
              list == null
                  ? null
                  : list.stream().filter(Objects::nonNull).findFirst().orElse(null));
}
 

import com.google.common.util.concurrent.JdkFutureAdapters; //导入依赖的package包/类
/**
 * Sends ARP Request as packet-out from the given port (node connector).
 *
 * @param senderAddress
 *            the addresses used in sender part of ARP packet
 * @param tpa
 *            the target protocol address, in this case IPv4 address for
 *            which MAC should be discovered
 * @param egressNc
 *            the path to node connector from where the ARP packet will be
 *            sent
 * @return future result about success of packet-out
 */
public ListenableFuture<RpcResult<Void>> sendArp(ArpMessageAddress senderAddress, Ipv4Address tpa,
		InstanceIdentifier<NodeConnector> egressNc) {
	checkNotNull(senderAddress);
	checkNotNull(tpa);
	checkNotNull(egressNc);
	final Ethernet arpFrame = createArpFrame(senderAddress, tpa);
	byte[] arpFrameAsBytes;
	try {
		arpFrameAsBytes = arpFrame.serialize();
	} catch (PacketException e) {
		LOG.warn("Serializition of ARP packet is not successful.", e);
		if (LOG.isDebugEnabled()) {
			LOG.debug("ARP packet: {}", ArpUtils.getArpFrameToStringFormat(arpFrame));
		}
		return Futures.immediateFailedFuture(e);
	}
	// Generate packet with destination switch and port

	TransmitPacketInput packet = new TransmitPacketInputBuilder().setEgress(new NodeConnectorRef(egressNc))
			.setNode(new NodeRef(egressNc.firstIdentifierOf(Node.class))).setPayload(arpFrameAsBytes).build();
	if (LOG.isTraceEnabled()) {
		LOG.trace("Sending ARP REQUEST \n{}", ArpUtils.getArpFrameToStringFormat(arpFrame));
	}
	Future<RpcResult<Void>> futureTransmitPacketResult = packetProcessingService.transmitPacket(packet);
	return JdkFutureAdapters.listenInPoolThread(futureTransmitPacketResult);
}
 

import com.google.common.util.concurrent.JdkFutureAdapters; //导入依赖的package包/类
/**
 * <p>executeRequest.</p>
 *
 * @param httpRequest a {@link org.apache.http.client.methods.HttpRequestBase} object.
 * @return a {@link org.apache.http.HttpResponse} object.
 * @throws java.io.IOException if any.
 */
public ListenableFuture<HttpResponse> executeRequest(
            final HttpRequest httpRequest)
        throws IOException {
    acquireRequestPermit();
    Future<HttpResponse> response = null;
    try {
        URL url = new URL(httpRequest.getRequestLine().getUri());
        HttpHost httphost =
            new HttpHost(url.getHost(),
                         ((url.getPort() == -1)
                             ? url.getDefaultPort()
                             : url.getPort()),
                         (url.toString().startsWith("https")
                             ? "https" : "http"));
        response = httpClient.execute(httphost, httpRequest, null);
    } catch (Exception e) {
        System.err.println("Exception occurred during asynchronous "
                           + "HTTP request: " + e.toString());
    }
    if (response == null) {
        return null;
    }
    ListenableFuture<HttpResponse> hr =
        JdkFutureAdapters.listenInPoolThread(response, executorService);
    return hr;
}
 

import com.google.common.util.concurrent.JdkFutureAdapters; //导入依赖的package包/类
@Override
public Future<RpcResult<Void>> resetSession(final ResetSessionInput input) {
    final ListenableFuture<?> f = this.peerRPCs.releaseConnection();
    return Futures.transform(JdkFutureAdapters.listenInPoolThread(f), input1 -> {
        if (f.isDone()) {
            return RpcResultBuilder.<Void>success().build();
        }
        return RpcResultBuilder.<Void>failed().withError(ErrorType.RPC, FAILURE_RESET_SESSION_MSG).build();
    }, MoreExecutors.directExecutor());
}
 

import com.google.common.util.concurrent.JdkFutureAdapters; //导入依赖的package包/类
@Override
public Future<RpcResult<Void>> routeRefreshRequest(final RouteRefreshRequestInput input) {
    final ChannelFuture f = sendRRMessage(input);
    if (f != null) {
        return Futures.transform(JdkFutureAdapters.listenInPoolThread(f), input1 -> {
            if (f.isSuccess()) {
                return RpcResultBuilder.<Void>success().build();
            }
            return RpcResultBuilder.<Void>failed().withError(ErrorType.RPC, FAILURE_MSG).build();
        }, MoreExecutors.directExecutor());
    }
    return RpcResultBuilder.<Void>failed().withError(ErrorType.RPC, FAILURE_MSG +
            " due to unsupported address families.").buildFuture();
}
 

import com.google.common.util.concurrent.JdkFutureAdapters; //导入依赖的package包/类
public void collectGetFeaturesOuput(Future<RpcResult<GetFeaturesOutput>> switchReply,
        final ConnectionAdapter connectionAdapter) {
    Futures.addCallback(JdkFutureAdapters.listenInPoolThread(switchReply),
            new FutureCallback<RpcResult<GetFeaturesOutput>>() {
                @Override
                public void onSuccess(RpcResult<GetFeaturesOutput> rpcFeatures) {
                    if (rpcFeatures.isSuccessful()) {
                        GetFeaturesOutput featureOutput = rpcFeatures.getResult();
                        // Register Switch connection/DatapathId to registry
                        connectionRegistry.registerConnectionAdapter(connectionAdapter, featureOutput);
                        NodeUpdated nodeUpdated = nodeAdded(connectionAdapter);
                        notificationProviderService.offerNotification(nodeUpdated);

                    } else {
                        // Handshake failed
                        for (RpcError rpcError : rpcFeatures.getErrors()) {
                            LOG.info("handshake - features failure [{}]: i:{} | m:{} | s:{}", rpcError.getInfo(),
                                    rpcError.getMessage(), rpcError.getSeverity(), rpcError.getCause());
                        }
                    }
                }

                @Override
                public void onFailure(Throwable t) {
                    LOG.info("getting feature failed seriously [addr:{}]: {}", connectionAdapter.getRemoteAddress(),
                            t.getMessage());
                }
            });
}
 

import com.google.common.util.concurrent.JdkFutureAdapters; //导入依赖的package包/类
private ListenableFuture<RpcResult<?>> invoke(final SchemaPath schemaPath, final DataObject input) {
    return JdkFutureAdapters.listenInPoolThread(invoker.invokeRpc(delegate, schemaPath.getLastComponent(), input));
}
 

import com.google.common.util.concurrent.JdkFutureAdapters; //导入依赖的package包/类
public ListenableFuture<List<String>> publishMessagesAsync(
		GroupedMessage groupedMessage) {
	return JdkFutureAdapters.listenInPoolThread(client
			.publishAsync(groupedMessage.getTopic(), groupedMessage.getMessages()));
}