org.elasticsearch.action.ActionListener.wrap()方法的使用及代码示例

x33g5p2x  于2022-01-16 转载在 其他  
字(12.9k)|赞(0)|评价(0)|浏览(160)

本文整理了Java中org.elasticsearch.action.ActionListener.wrap()方法的一些代码示例，展示了ActionListener.wrap()的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。ActionListener.wrap()方法的具体详情如下：
包路径：org.elasticsearch.action.ActionListener
类名称：ActionListener
方法名：wrap

ActionListener.wrap介绍

[英]Creates a listener that listens for a response (or failure) and executes the corresponding runnable when the response (or failure) is received.
[中]创建侦听响应（或故障）的侦听器，并在收到响应（或故障）时执行相应的runnable。

代码示例

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. /**
  2.  * Creates a listener that listens for a response (or failure) and executes the
  3.  * corresponding runnable when the response (or failure) is received.
  4.  *
  5.  * @param runnable the runnable that will be called in event of success or failure
  6.  * @param <Response> the type of the response
  7.  * @return a listener that listens for responses and invokes the runnable when received
  8.  */
  9. static <Response> ActionListener<Response> wrap(Runnable runnable) {
  10.   return wrap(r -> runnable.run(), e -> runnable.run());
  11. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. void updateRemoteCluster(
  2.     final String clusterAlias,
  3.     final List<String> addresses,
  4.     final String proxyAddress,
  5.     final ActionListener<Void> connectionListener) {
  6.   final List<Tuple<String, Supplier<DiscoveryNode>>> nodes =
  7.       addresses.stream().<Tuple<String, Supplier<DiscoveryNode>>>map(address -> Tuple.tuple(address, () ->
  8.           buildSeedNode(clusterAlias, address, Strings.hasLength(proxyAddress)))
  9.       ).collect(Collectors.toList());
  10.   updateRemoteClusters(Collections.singletonMap(clusterAlias, new Tuple<>(proxyAddress, nodes)), connectionListener);
  11. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. protected void serverAcceptedChannel(TcpChannel channel) {
  2.   boolean addedOnThisCall = acceptedChannels.add(channel);
  3.   assert addedOnThisCall : "Channel should only be added to accepted channel set once";
  4.   // Mark the channel init time
  5.   channel.getChannelStats().markAccessed(threadPool.relativeTimeInMillis());
  6.   channel.addCloseListener(ActionListener.wrap(() -> acceptedChannels.remove(channel)));
  7.   logger.trace(() -> new ParameterizedMessage("Tcp transport channel accepted: {}", channel));
  8. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. /**
  2.  * Fetches all shards for the search request from this remote connection. This is used to later run the search on the remote end.
  3.  */
  4. public void fetchSearchShards(ClusterSearchShardsRequest searchRequest,
  5.                ActionListener<ClusterSearchShardsResponse> listener) {
  6.   final ActionListener<ClusterSearchShardsResponse> searchShardsListener;
  7.   final Consumer<Exception> onConnectFailure;
  8.   if (skipUnavailable) {
  9.     onConnectFailure = (exception) -> listener.onResponse(ClusterSearchShardsResponse.EMPTY);
  10.     searchShardsListener = ActionListener.wrap(listener::onResponse, (e) -> listener.onResponse(ClusterSearchShardsResponse.EMPTY));
  11.   } else {
  12.     onConnectFailure = listener::onFailure;
  13.     searchShardsListener = listener;
  14.   }
  15.   // in case we have no connected nodes we try to connect and if we fail we either notify the listener or not depending on
  16.   // the skip_unavailable setting
  17.   ensureConnected(ActionListener.wrap((x) -> fetchShardsInternal(searchRequest, searchShardsListener), onConnectFailure));
  18. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. @Override
  2.   protected void doExecute(RemoteInfoRequest remoteInfoRequest, ActionListener<RemoteInfoResponse> listener) {
  3.     remoteClusterService.getRemoteConnectionInfos(ActionListener.wrap(remoteConnectionInfos
  4.         -> listener.onResponse(new RemoteInfoResponse(remoteConnectionInfos)), listener::onFailure));
  5.   }
  6. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. /**
  2.  * Notifies the master node to create new persistent task and to assign it to a node.
  3.  */
  4. public <Params extends PersistentTaskParams> void sendStartRequest(final String taskId,
  5.                                   final String taskName,
  6.                                   final Params taskParams,
  7.                                   final ActionListener<PersistentTask<Params>> listener) {
  8.   @SuppressWarnings("unchecked")
  9.   final ActionListener<PersistentTask<?>> wrappedListener =
  10.     ActionListener.wrap(t -> listener.onResponse((PersistentTask<Params>) t), listener::onFailure);
  11.   StartPersistentTaskAction.Request request = new StartPersistentTaskAction.Request(taskId, taskName, taskParams);
  12.   execute(request, StartPersistentTaskAction.INSTANCE, wrappedListener);
  13. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. public final void run() {
  2.   final ScrollIdForNode[] context = scrollId.getContext();
  3.   if (context.length == 0) {
  4.     listener.onFailure(new SearchPhaseExecutionException("query", "no nodes to search on", ShardSearchFailure.EMPTY_ARRAY));
  5.   } else {
  6.     collectNodesAndRun(Arrays.asList(context), nodes, searchTransportService, ActionListener.wrap(lookup -> run(lookup, context),
  7.       listener::onFailure));
  8.   }
  9. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. ActionListener<BulkResponse> wrapActionListenerIfNeeded(long ingestTookInMillis, ActionListener<BulkResponse> actionListener) {
  2.   if (itemResponses.isEmpty()) {
  3.     return ActionListener.wrap(
  4.         response -> actionListener.onResponse(new BulkResponse(response.getItems(),
  5.             response.getTook().getMillis(), ingestTookInMillis)),
  6.         actionListener::onFailure);
  7.   } else {
  8.     return new IngestBulkResponseListener(ingestTookInMillis, originalSlots, itemResponses, actionListener);
  9.   }
  10. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. /**
  2.  * Executes an asynchronous persistent task action using the client.
  3.  * <p>
  4.  * The origin is set in the context and the listener is wrapped to ensure the proper context is restored
  5.  */
  6. private <Req extends ActionRequest, Resp extends PersistentTaskResponse, Builder extends ActionRequestBuilder<Req, Resp, Builder>>
  7.   void execute(final Req request, final Action<Req, Resp, Builder> action, final ActionListener<PersistentTask<?>> listener) {
  8.     try {
  9.       client.execute(action, request,
  10.           ActionListener.wrap(r -> listener.onResponse(r.getTask()), listener::onFailure));
  11.     } catch (Exception e) {
  12.       listener.onFailure(e);
  13.     }
  14. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. void registerNodeConnection(List<TcpChannel> nodeChannels, ConnectionProfile connectionProfile) {
  2.   TimeValue pingInterval = connectionProfile.getPingInterval();
  3.   if (pingInterval.millis() < 0) {
  4.     return;
  5.   }
  6.   final ScheduledPing scheduledPing = pingIntervals.computeIfAbsent(pingInterval, ScheduledPing::new);
  7.   scheduledPing.ensureStarted();
  8.   for (TcpChannel channel : nodeChannels) {
  9.     scheduledPing.addChannel(channel);
  10.     channel.addCloseListener(ActionListener.wrap(() -> {
  11.       scheduledPing.removeChannel(channel);
  12.     }));
  13.   }
  14. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. public void openIndex(final OpenIndexClusterStateUpdateRequest request,
  2.            final ActionListener<OpenIndexClusterStateUpdateResponse> listener) {
  3.   onlyOpenIndex(request, ActionListener.wrap(response -> {
  4.     if (response.isAcknowledged()) {
  5.       String[] indexNames = Arrays.stream(request.indices()).map(Index::getName).toArray(String[]::new);
  6.       activeShardsObserver.waitForActiveShards(indexNames, request.waitForActiveShards(), request.ackTimeout(),
  7.         shardsAcknowledged -> {
  8.           if (shardsAcknowledged == false) {
  9.             logger.debug("[{}] indices opened, but the operation timed out while waiting for " +
  10.               "enough shards to be started.", Arrays.toString(indexNames));
  11.           }
  12.           listener.onResponse(new OpenIndexClusterStateUpdateResponse(response.isAcknowledged(), shardsAcknowledged));
  13.         }, listener::onFailure);
  14.     } else {
  15.       listener.onResponse(new OpenIndexClusterStateUpdateResponse(false, false));
  16.     }
  17.   }, listener::onFailure));
  18. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. @Override
  2.   protected QueryBuilder doRewrite(QueryRewriteContext queryRewriteContext) {
  3.     if (supplier != null) {
  4.       return supplier.get() == null ? this : new TermsQueryBuilder(this.fieldName, supplier.get());
  5.     } else if (this.termsLookup != null) {
  6.       SetOnce<List<?>> supplier = new SetOnce<>();
  7.       queryRewriteContext.registerAsyncAction((client, listener) -> {
  8.         fetch(termsLookup, client, ActionListener.wrap(list -> {
  9.           supplier.set(list);
  10.           listener.onResponse(null);
  11.         }, listener::onFailure));
  13.       });
  14.       return new TermsQueryBuilder(this.fieldName, supplier::get);
  15.     }
  16.     return this;
  17.   }
  18. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. @Override
  2. protected void masterOperation(final ClusterRerouteRequest request, final ClusterState state,
  3.                 final ActionListener<ClusterRerouteResponse> listener) {
  4.   ActionListener<ClusterRerouteResponse> logWrapper = ActionListener.wrap(
  5.     response -> {
  6.       if (request.dryRun() == false) {
  7.         response.getExplanations().getYesDecisionMessages().forEach(logger::info);
  8.       }
  9.       listener.onResponse(response);
  10.     },
  11.     listener::onFailure
  12.   );
  13.   clusterService.submitStateUpdateTask("cluster_reroute (api)", new ClusterRerouteResponseAckedClusterStateUpdateTask(logger,
  14.     allocationService, request, logWrapper));
  15. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. @Override
  2. public void onResponse(Version version) {
  3.   NodeChannels nodeChannels = new NodeChannels(node, channels, connectionProfile, version);
  4.   long relativeMillisTime = threadPool.relativeTimeInMillis();
  5.   nodeChannels.channels.forEach(ch -> {
  6.     // Mark the channel init time
  7.     ch.getChannelStats().markAccessed(relativeMillisTime);
  8.     ch.addCloseListener(ActionListener.wrap(nodeChannels::close));
  9.   });
  10.   keepAlive.registerNodeConnection(nodeChannels.channels, connectionProfile);
  11.   listener.onResponse(nodeChannels);
  12. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. public void updateGlobalCheckpointForShard(final ShardId shardId) {
  2.   final ThreadContext threadContext = threadPool.getThreadContext();
  3.   try (ThreadContext.StoredContext ignore = threadContext.stashContext()) {
  4.     threadContext.markAsSystemContext();
  5.     execute(
  6.         new Request(shardId),
  7.         ActionListener.wrap(r -> {
  8.         }, e -> {
  9.           if (ExceptionsHelper.unwrap(e, AlreadyClosedException.class, IndexShardClosedException.class) == null) {
  10.             logger.info(new ParameterizedMessage("{} global checkpoint sync failed", shardId), e);
  11.           }
  12.         }));
  13.   }
  14. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. @Override
  2. protected <Request extends ActionRequest, Response extends ActionResponse, RequestBuilder extends
  3.   ActionRequestBuilder<Request, Response, RequestBuilder>>
  4. void doExecute(Action<Request, Response, RequestBuilder> action, Request request, ActionListener<Response> listener) {
  5.   remoteClusterService.ensureConnected(clusterAlias, ActionListener.wrap(res -> {
  6.     Transport.Connection connection = remoteClusterService.getConnection(clusterAlias);
  7.     service.sendRequest(connection, action.name(), request, TransportRequestOptions.EMPTY,
  8.       new ActionListenerResponseHandler<>(listener, action.getResponseReader()));
  9.   },
  10.   listener::onFailure));
  11. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. void cleanScrollIds(List<ScrollIdForNode> parsedScrollIds) {
  2.   SearchScrollAsyncAction.collectNodesAndRun(parsedScrollIds, nodes, searchTransportService, ActionListener.wrap(
  3.     lookup -> {
  4.       for (ScrollIdForNode target : parsedScrollIds) {
  5.         final DiscoveryNode node = lookup.apply(target.getClusterAlias(), target.getNode());
  6.         if (node == null) {
  7.           onFreedContext(false);
  8.         } else {
  9.           try {
  10.             Transport.Connection connection = searchTransportService.getConnection(target.getClusterAlias(), node);
  11.             searchTransportService.sendFreeContext(connection, target.getScrollId(),
  12.               ActionListener.wrap(freed -> onFreedContext(freed.isFreed()), e -> onFailedFreedContext(e, node)));
  13.           } catch (Exception e) {
  14.             onFailedFreedContext(e, node);
  15.           }
  16.         }
  17.       }
  18.     }, listener::onFailure));
  19. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. public static <Response extends ReplicationResponse & WriteResponse>
  2. ActionListener<BulkResponse> wrapBulkResponse(ActionListener<Response> listener) {
  3.   return ActionListener.wrap(bulkItemResponses -> {
  4.     assert bulkItemResponses.getItems().length == 1 : "expected only one item in bulk request";
  5.     BulkItemResponse bulkItemResponse = bulkItemResponses.getItems()[0];
  6.     if (bulkItemResponse.isFailed() == false) {
  7.       final DocWriteResponse response = bulkItemResponse.getResponse();
  8.       listener.onResponse((Response) response);
  9.     } else {
  10.       listener.onFailure(bulkItemResponse.getFailure().getCause());
  11.     }
  12.   }, listener::onFailure);
  13. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. @Override
  2. public void onResponse(IndicesStatsResponse indicesStatsResponse) {
  3.   CreateIndexClusterStateUpdateRequest updateRequest = prepareCreateIndexRequest(resizeRequest, state,
  4.     (i) -> {
  5.       IndexShardStats shard = indicesStatsResponse.getIndex(sourceIndex).getIndexShards().get(i);
  6.       return shard == null ? null : shard.getPrimary().getDocs();
  7.     }, sourceIndex, targetIndex);
  8.   createIndexService.createIndex(
  9.     updateRequest,
  10.     ActionListener.wrap(response ->
  11.         listener.onResponse(new ResizeResponse(response.isAcknowledged(), response.isShardsAcknowledged(),
  12.             updateRequest.index())), listener::onFailure
  13.     )
  14.   );
  15. }

代码示例来源：origin: org.elasticsearch/elasticsearch

  1. @Override
  2. protected void masterOperation(final CreateIndexRequest request, final ClusterState state,
  3.                 final ActionListener<CreateIndexResponse> listener) {
  4.   String cause = request.cause();
  5.   if (cause.length() == 0) {
  6.     cause = "api";
  7.   }
  8.   final String indexName = indexNameExpressionResolver.resolveDateMathExpression(request.index());
  9.   final CreateIndexClusterStateUpdateRequest updateRequest =
  10.     new CreateIndexClusterStateUpdateRequest(request, cause, indexName, request.index(), request.updateAllTypes())
  11.       .ackTimeout(request.timeout()).masterNodeTimeout(request.masterNodeTimeout())
  12.       .settings(request.settings()).mappings(request.mappings())
  13.       .aliases(request.aliases())
  14.       .waitForActiveShards(request.waitForActiveShards());
  15.   createIndexService.createIndex(updateRequest, ActionListener.wrap(response ->
  16.     listener.onResponse(new CreateIndexResponse(response.isAcknowledged(), response.isShardsAcknowledged(), indexName)),
  17.     listener::onFailure));
  18. }

内容来源于网络，如有侵权，请联系作者删除！

ActionListener

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com