Java 类io.vertx.core.shareddata.AsyncMap 实例源码

/**
 * Create a AsyncMap
 *
 * @param <K> Key type
 * @param <V> Value type
 * @param vertx
 * @param mapName name of the map. If null, will always create a local map
 * @param fut
 */
public static <K, V> void create(Vertx vertx, String mapName,          Handler<ExtendedAsyncResult<AsyncMap<K, V>>> fut) {
  if (vertx.isClustered() && mapName != null) {
    SharedData shared = vertx.sharedData();
    shared.<K, V>getClusterWideMap(mapName, res -> {
      if (res.succeeded()) {
        fut.handle(new Success<>(res.result()));
      } else {
        fut.handle(new Failure<>(INTERNAL, res.cause()));
      }
    });
  } else {
    // Dirty trickery to make sure we can run two verticles in our tests,
    // without them sharing the 'shared' memory. Only when running in non-
    // clustered mode, of course.
    // Also used in deploy-only nodes, where we want local-only tenant and
    // module lists with only the hard-coded supertenant and internalModule.
    Random r = new Random();
    String newid = String.format("%09d", r.nextInt(1000000000));
    if (mapName != null) {
      newid = mapName + newid;
    }
    AsyncLocalmap<K, V> l = new AsyncLocalmap<>(vertx, newid);
    fut.handle(new Success<>(l));
  }
}

private synchronized void retrieveRegistry(Handler<AsyncResult<AsyncMap<String, String>>> handler) {
  if (registry != null) {
    handler.handle(Future.succeededFuture(registry));
  } else {
    vertx.sharedData().<String, String>getClusterWideMap("service.registry", ar -> {
      synchronized (DefaultServiceDiscoveryBackend.class) {
        if (ar.failed()) {
          handler.handle(ar);
        } else {
          registry = ar.result();
          handler.handle(Future.succeededFuture(registry));
        }
      }
    });
  }
}

private void getStore(Handler<AsyncResult<AsyncMap<String, Buffer>>> handler) {
  if (store != null) {
    handler.handle(Future.succeededFuture(store));
    return;
  }

  String name = getClass().getName() + ".STORE";
  if (vertx.isClustered()) {
    vertx.sharedData().<String, Buffer>getClusterWideMap(name, ar -> {
      if (ar.succeeded()) {
        store = ar.result();
      }
      handler.handle(ar);
    });
  } else {
    LocalMap<String, Buffer> lm = vertx.sharedData().getLocalMap(name);
    store = new AsyncLocalMap<String, Buffer>(lm);
    handler.handle(Future.succeededFuture(store));
  }
}

private void doDeleteChunks(Queue<String> paths, AsyncMap<String, Buffer> store,
    Handler<AsyncResult<Void>> handler) {
  if (paths.isEmpty()) {
    handler.handle(Future.succeededFuture());
    return;
  }

  String path = PathUtils.normalize(paths.poll());
  store.remove(path, ar -> {
    if (ar.failed()) {
      handler.handle(Future.failedFuture(ar.cause()));
    } else {
      doDeleteChunks(paths, store, handler);
    }
  });
}

@Override
public <K, V> void getAsyncMap(final String name, final Handler<AsyncResult<AsyncMap<K, V>>> resultHandler) {
    ConcurrentMap map = asyncMaps.get(name);
    if (map == null) {
        map = new ConcurrentHashMap<>();
        final ConcurrentMap prevMap = asyncMaps.putIfAbsent(name, map);
        if (prevMap != null) {
            map = prevMap;
        }
    }
    @SuppressWarnings("unchecked") final ConcurrentMap<K, V> theMap = map;
    this.vertx.runOnContext(v -> resultHandler.handle(Future.succeededFuture(new FakeAsyncMap<>(theMap))));
}

@SuppressWarnings("unchecked")
private synchronized void adaptListener() {
    if (listenerCleaner == null) {
        // TODO - move in separated jar as some sort of provider
        AsyncMap<String, Session> map = Reflection.field("sessionMap").ofType(AsyncMap.class).in(sessionStore).get();
        String listenerId = tryGetHazelcastMap(map)
            .map(imap -> imap.addEntryListener(new MapListenerAdapter<String, Session>() {
                @Override
                public void entryExpired(EntryEvent<String, Session> event) {
                    sessionExpiredProducer.send(event.getKey());
                }
            }, true)).orElse(null);
        listenerCleaner = () -> tryGetHazelcastMap(map).ifPresent(imap -> imap.removeEntryListener(listenerId));
    }
}

private Optional<IMap> tryGetHazelcastMap(AsyncMap<String, Session> map) {
    return Optional.ofNullable(map)
        .map( m -> Reflection.field("delegate").ofType(AsyncMap.class).in(m).get())
        .filter(HazelcastAsyncMap.class::isInstance)
        .map(HazelcastAsyncMap.class::cast)
        .map(h -> Reflection.field("map").ofType(IMap.class).in(h).get());
}

@Override
public <K, V> void getAsyncMap(String name, Handler<AsyncResult<AsyncMap<K, V>>> resultHandler) {
  vertx.executeBlocking(future -> {
    Cache<Object, Object> cache = cacheManager.getCache(name);
    future.complete(new InfinispanAsyncMapImpl<>(vertx, cache));
  }, false, resultHandler);
}

/**
 * Unwraps a generic {@link AsyncMap} to an {@link InfinispanAsyncMap}.
 *
 * @throws IllegalArgumentException if underlying implementation is not Infinispan
 */
@SuppressWarnings("unchecked")
static <K, V> InfinispanAsyncMap<K, V> unwrap(AsyncMap asyncMap) {
  if (asyncMap instanceof WrappedAsyncMap) {
    WrappedAsyncMap wrappedAsyncMap = (WrappedAsyncMap) asyncMap;
    AsyncMap delegate = wrappedAsyncMap.getDelegate();
    if (delegate instanceof InfinispanAsyncMap) {
      return (InfinispanAsyncMap<K, V>) delegate;
    }
  }
  throw new IllegalArgumentException(String.valueOf(asyncMap != null ? asyncMap.getClass() : null));
}

@Override
public <K, V> void getAsyncMap(String name, Handler<AsyncResult<AsyncMap<K, V>>> handler) {
  atomix.<K, V>consistentMapBuilder(name)
      .withSerializer(createSerializer())
      .withConsistency(Consistency.LINEARIZABLE)
      .withPersistence(Persistence.PERSISTENT)
      .withReplication(Replication.SYNCHRONOUS)
      .withRecovery(Recovery.RECOVER)
      .withMaxRetries(5)
      .buildAsync()
      .whenComplete(VertxFutures.convertHandler(
          handler, map -> new AtomixAsyncMap<>(vertx, map.async()), vertx.getOrCreateContext()));
}

@Override
public <K, V> void getAsyncMap(String name, Handler<AsyncResult<AsyncMap<K, V>>> handler) {
  logTrace(() -> String.format("Create new AsyncMap [%s] on address [%s]", name, address));
  vertx.executeBlocking((future) -> {
    checkCluster();
    AsyncMap<K, V> map = cacheManager.<K, V>createAsyncMap(name);
    future.complete(map);
  }, handler);
}

private void getEndpointHolderAndAdd(ServerWebSocket serverSocket, AsyncMap<String, WSEndpointHolder> registryMap) {
    registryMap.get(WS_ENDPOINT_HOLDER, wsEndpointHolder -> {
        if (wsEndpointHolder.succeeded()) {
            updateWSEndpointHolder(serverSocket, registryMap, wsEndpointHolder);
        }
    });

}

private void updateWSEndpointHolder(ServerWebSocket serverSocket, AsyncMap<String, WSEndpointHolder> registryMap, AsyncResult<WSEndpointHolder> wsEndpointHolder) {
    log("add entry: " + Thread.currentThread());
    final String binaryHandlerId = serverSocket.binaryHandlerID();
    final String textHandlerId = serverSocket.textHandlerID();
    final String path = serverSocket.path();
    final EventBus eventBus = vertx.eventBus();
    final WSEndpoint endpoint = new WSEndpoint(binaryHandlerId, textHandlerId, path);
    final WSEndpointHolder result = wsEndpointHolder.result();
    if (result != null) {
        addDefinitionToRegistry(serverSocket, eventBus, path, endpoint, registryMap, result);
    } else {
        createEntryAndAddDefinition(serverSocket, eventBus, path, endpoint, registryMap);
    }
}

private void createEntryAndAddDefinition(ServerWebSocket serverSocket, EventBus eventBus, String path, WSEndpoint endpoint, AsyncMap<String, WSEndpointHolder> registryMap) {
    final WSEndpointHolder holder = new WSEndpointHolder();
    holder.add(endpoint);
    registryMap.put(WS_ENDPOINT_HOLDER, holder, s -> {
                if (s.succeeded()) {
                    log("OK ADD: " + serverSocket.binaryHandlerID() + "  Thread" + Thread.currentThread());
                    sendToWSService(serverSocket, eventBus, path, endpoint);
                }
            }

    );
}

private void addDefinitionToRegistry(ServerWebSocket serverSocket, EventBus eventBus, String path, WSEndpoint endpoint, AsyncMap<String, WSEndpointHolder> registryMap, WSEndpointHolder wsEndpointHolder) {
    wsEndpointHolder.add(endpoint);
    registryMap.replace(WS_ENDPOINT_HOLDER, wsEndpointHolder, s -> {
                if (s.succeeded()) {
                    log("OK REPLACE: " + serverSocket.binaryHandlerID() + "  Thread" + Thread.currentThread());
                    sendToWSService(serverSocket, eventBus, path, endpoint);
                }
            }
    );
}

private void retrieveEndpointHolderAndRemove(ServerWebSocket serverSocket, String binaryHandlerID, String textHandlerID, AsyncMap<String, WSEndpointHolder> registryMap, AsyncResult<WSEndpointHolder> wsEndpointHolder) {
    if (wsEndpointHolder.succeeded()) {
        final WSEndpointHolder result = wsEndpointHolder.result();
        if (result != null) {
            findEndpointAndRemove(serverSocket, binaryHandlerID, textHandlerID, registryMap,result);

        }
    }
}

private void findEndpointAndRemove(ServerWebSocket serverSocket, String binaryHandlerID, String textHandlerID, AsyncMap<String, WSEndpointHolder> registryMap, WSEndpointHolder wsEndpointHolder) {
    final List<WSEndpoint> all = wsEndpointHolder.getAll();
    final Optional<WSEndpoint> first = all.stream().filter(e -> e.getBinaryHandlerId().equals(binaryHandlerID) && e.getTextHandlerId().equals(textHandlerID)).findFirst();
    if (first.isPresent()) {
        first.ifPresent(endpoint -> {
            wsEndpointHolder.remove(endpoint);
            registryMap.replace(WS_ENDPOINT_HOLDER, wsEndpointHolder, replaceHolder -> log("OK REMOVE: " + serverSocket.binaryHandlerID() + "  succeed:" + replaceHolder.succeeded()));
        });
    }
}

private void getServiceHolderAndReplyToServiceInfoRequest(Message<byte[]> message, AsyncMap<String, ServiceInfoHolder> resultMap) {
    resultMap.get(GlobalKeyHolder.SERVICE_HOLDER, onSuccess(resultHolder -> {
        if (resultHolder != null) {
            message.reply(getServiceIfoHolderBinary(buildServiceInfoForEntryPoint(resultHolder)));
        } else {
            message.reply(getServiceIfoHolderBinary(new ServiceInfoHolder()));
        }
    }));
}

private void getServiceHolderAndRegister(Message<byte[]> message, ServiceInfo info, AsyncMap<String, ServiceInfoHolder> resultMap) {
    logDebug("got map");
    resultMap.get(GlobalKeyHolder.SERVICE_HOLDER, onSuccess(resultHolder -> {
        logDebug("got result holder");
        if (resultHolder != null) {
            addServiceEntry(resultMap, info, resultHolder, message);
        } else {
            createNewEntry(resultMap, info, new ServiceInfoHolder(), message);
        }
    }));
}

private void addServiceEntry(final AsyncMap resultMap, final ServiceInfo info, final ServiceInfoHolder holder, final Message<byte[]> message) {
    holder.add(info);
    logDebug("update result holder");
    resultMap.replace(GlobalKeyHolder.SERVICE_HOLDER, holder, onSuccess(s -> {
        publishToEntryPoint(info);
        message.reply(true);
        logDebug("Register REPLACE: " + info);
    }));
}

private void createNewEntry(final AsyncMap resultMap, final ServiceInfo info, final ServiceInfoHolder holder, final Message<byte[]> message) {
    holder.add(info);
    logDebug("add result holder");
    resultMap.put(GlobalKeyHolder.SERVICE_HOLDER, holder, onSuccess(s -> {
        publishToEntryPoint(info);
        message.reply(true);
        logDebug("Register ADD: " + info);
    }));
}

private void getServiceHolderAndPingServices(AsyncMap<String, ServiceInfoHolder> resultMap) {
    resultMap.get(GlobalKeyHolder.SERVICE_HOLDER, onSuccess(holder -> {
        logDebug("get Holder " + holder + " this:" + this);
        if (holder != null) {
            final List<ServiceInfo> serviceHolders = holder.getAll();
            serviceHolders.forEach(this::pingService);

        }

    }));
}

private void removeAndUpdateServiceInfo(ServiceInfo info, AsyncMap<String, ServiceInfoHolder> resultMap, ServiceInfoHolder holder) {
    holder.remove(info);
    resultMap.replace(GlobalKeyHolder.SERVICE_HOLDER, holder, t -> {
        if (t.succeeded()) {
            resetServiceCounterAndPublish(info);

        }
    });
}

private void getMap(Handler<AsyncResult<AsyncMap<String, Session>>> resultHandler) {
  if (sessionMap == null) {
    vertx.sharedData().<String, Session>getClusterWideMap(sessionMapName, res -> {
      if (res.succeeded()) {
        sessionMap = res.result();
        resultHandler.handle(Future.succeededFuture(res.result()));
      } else {
        resultHandler.handle(res);
      }
    });
  } else {
    resultHandler.handle(Future.succeededFuture(sessionMap));
  }
}

@Override
public <K, V> void getAsyncMap(String name, Handler<AsyncResult<AsyncMap<K, V>>> resultHandler) {
  vertx.executeBlocking(fut -> {
    IMap<K, V> map = hazelcast.getMap(name);
    fut.complete(USE_HZ_ASYNC_API ? new HazelcastInternalAsyncMap<>(vertx, map) : new HazelcastAsyncMap<>(vertx, map));
  }, resultHandler);
}

@Override
public <K, V> void getAsyncMap(String name, Handler<AsyncResult<AsyncMap<K, V>>> handler) {
  vertx.runOnContext(event -> handler.handle(Future.succeededFuture(new ZKAsyncMap<>(vertx, curator, name))));
}

@Override
public <K, V> void getAsyncMap(String name, Handler<AsyncResult<AsyncMap<K, V>>> handler) {
  vertx.executeBlocking(
    fut -> fut.complete(new AsyncMapImpl<>(getCache(name), vertx)), handler
  );
}

public <K, V> void asyncMapStreams(AsyncMap<K, V> asyncMap) {
  InfinispanAsyncMap<K, V> infinispanAsyncMap = InfinispanAsyncMap.unwrap(asyncMap);
  ReadStream<K> keyStream = infinispanAsyncMap.keyStream();
  ReadStream<V> valueStream = infinispanAsyncMap.valueStream();
  ReadStream<Map.Entry<K, V>> entryReadStream = infinispanAsyncMap.entryStream();
}

public <K, V> AsyncMap<K, V> createAsyncMap(String name) {
    logDebug(() -> String.format("method createAsyncMap address[%s] name[%s]", channel.getAddressAsString(), name));
    Map<K, V> map = mapService.<K, V>mapCreate(name);
    return new AsyncMapWrapper<>(name, map, executorService);
}