Java 类io.vertx.core.impl.VertxInternal 实例源码

@Override
public Object execute() throws Exception {

    ShellTable table = new ShellTable();

    table.column("ID");
    table.column("Identifier");
    table.column("Options");

    getVertxService().deploymentIDs().forEach(id -> { 
            Deployment deployment = ((VertxInternal)getVertxService()).getDeployment(id);
            Row row = table.addRow();
            row.addContent(id, deployment.verticleIdentifier(), deployment.deploymentOptions().toJson());
        }
    );

    try {
        table.print(System.out);
    } catch (Throwable t)  {
        System.err.println("FAILED to write table");
    }

    return null;
}

private EmfProvidereHolder lookupHolder() {

    synchronized (vertx) {
      LocalMap<String, EmfProvidereHolder> map = vertx.sharedData().getLocalMap(DS_LOCAL_MAP_NAME);
      EmfProvidereHolder theHolder = map.get(datasourceName);
      if (theHolder == null) {
        theHolder = new EmfProvidereHolder((VertxInternal) vertx, config, map, datasourceName);
      } else {
        theHolder.incRefCount();
      }
      return theHolder;
    }

  }

public EmfProvidereHolder(VertxInternal vertx, JsonObject config, LocalMap<String, EmfProvidereHolder> map, String name) {
  this.name = name;
  this.config = config;
  this.vertx = vertx;
  this.map = map;
  this.emfProvider = initializeProvider().setConfig(config).setVertx(vertx);
}

CircuitBreakerMetrics(Vertx vertx, CircuitBreakerImpl circuitBreaker, CircuitBreakerOptions options) {
  this.circuitBreaker = circuitBreaker;
  this.circuitBreakerTimeout = circuitBreaker.options().getTimeout();
  this.circuitBreakerResetTimeout = circuitBreaker.options().getResetTimeout();
  this.node = vertx.isClustered() ? ((VertxInternal) vertx).getClusterManager().getNodeID() : "local";
  this.rollingWindow = new RollingWindow(options.getMetricsRollingWindow(), options.getMetricsRollingBuckets());
}

private String getNodeId(Vertx vertx) {
  if (vertx.isClustered()) {
    return ((VertxInternal) vertx).getNodeID();
  } else {
    return "localhost";
  }
}

@Before
public void setUp() {
  Vertx.clusteredVertx(new VertxOptions(), ar -> {
    ((VertxInternal) ar.result()).getClusterManager().<String, String>getAsyncMap("some-name", x -> {
      map = x.result();
      vertx = ar.result();
    });
  });
  await().until(() -> vertx != null);
}

/**
 * Registers a check which succeeds if a connection with the configured <em>Authentication</em> service can be established.
 *
 * @param readinessHandler The health check handler to register the checks with.
 */
@Override
public void registerReadinessChecks(final HealthCheckHandler readinessHandler) {

    if (dnsClient != null) {
        log.info("registering readiness check using DNS Client");
        readinessHandler.register("authentication-service-availability", status -> {
            log.trace("checking availability of Authentication service");
            dnsClient.lookup(getConfig().getHost(), lookupAttempt -> {
                if (lookupAttempt.succeeded()) {
                    status.tryComplete(Status.OK());
                } else {
                    log.debug("readiness check failed to resolve Authentication service address [{}]: ",
                            getConfig().getHost(), lookupAttempt.cause().getMessage());
                    status.tryComplete(Status.KO());
                }
            });
        });
    } else if (VertxInternal.class.isInstance(vertx)) {
        log.info("registering readiness check using vert.x Address Resolver");
        readinessHandler.register("authentication-service-availability", status -> {
            log.trace("checking availability of Authentication service");
            ((VertxInternal) vertx).resolveAddress(getConfig().getHost(), lookupAttempt -> {
                if (lookupAttempt.succeeded()) {
                    status.tryComplete(Status.OK());
                } else {
                    log.debug("readiness check failed to resolve Authentication service address [{}]: ",
                            getConfig().getHost(), lookupAttempt.cause().getMessage());
                    status.tryComplete(Status.KO());
                };
            });
        });
    } else {
        log.warn("cannot register readiness check, no DNS resolver available");
    }
}

private static String getNodeId(Vertx vertx) {
    if (vertx.isClustered()) {
        return ((VertxInternal) vertx).getNodeID();
    } else {
        return "localhost";
    }
}

/**
 * Constructor.
 *
 * @param cache {@link IgniteCache} instance.
 * @param vertx {@link Vertx} instance.
 */
public AsyncMultiMapImpl(IgniteCache<K, Set<V>> cache, Vertx vertx) {
  cache.unwrap(Ignite.class).events().localListen((IgnitePredicate<Event>)event -> {
    if (!(event instanceof CacheEvent)) {
      throw new IllegalArgumentException("Unknown event received: " + event);
    }

    CacheEvent cacheEvent = (CacheEvent)event;

    if (Objects.equals(cacheEvent.cacheName(), cache.getName()) &&
        ((IgniteCacheProxy)cache).context().localNodeId().equals(cacheEvent.eventNode().id())) {
      K key = unmarshal(cacheEvent.key());

      switch (cacheEvent.type()) {
        case EVT_CACHE_OBJECT_REMOVED:
          subs.remove(key);
          break;

        default:
          throw new IllegalArgumentException("Unknown event received: " + event);
      }
    }

    return true;
  }, EVT_CACHE_OBJECT_REMOVED);

  this.cache = cache;
  this.vertx = (VertxInternal) vertx;
}

public InfinispanAsyncMultiMap(Vertx vertx, Cache<MultiMapKey, Object> cache) {
  this.vertx = (VertxInternal) vertx;
  this.cache = cache;
  nearCache = new ConcurrentHashMap<>();
  cache.addListener(new EntryListener());
  taskQueue = new TaskQueue();
}

private void sendUpdateOnEventBus() {
  vertx.eventBus().publish("circuit-breaker[" + name + "]", new JsonObject()
      .put("name", name)
      .put("state", state)
      .put("failures", failures)
      .put("node", vertx.isClustered() ? ((VertxInternal) vertx).getClusterManager().getNodeID() : "local"));
}

public ExtendedAsyncMapImpl(Vertx vertx, String name) {
  this.vertx = vertx;
  ClusterManager clusterManager = ((VertxInternal) vertx).getClusterManager();
  if (clusterManager == null) {
    syncMap = new ConcurrentHashMap<>();
  } else {
    syncMap = clusterManager.getSyncMap(name);
  }
}

/**
 * Get a local counter. The counter will be passed to the handler.
 *
 * @param name the name of the counter.
 * @param resultHandler the handler
 */
public void getCounter(String name, Handler<AsyncResult<Counter>> resultHandler) {
  Objects.requireNonNull(name, "name");
  Objects.requireNonNull(resultHandler, "resultHandler");
  Counter counter = this.localCounters
      .computeIfAbsent(name, (n) -> new AsynchronousCounter((VertxInternal) this.vertx));
  Context context = this.vertx.getOrCreateContext();
  context.runOnContext((v) -> resultHandler.handle(Future.succeededFuture(counter)));
}

public CommandRegistryImpl(VertxInternal vertx) {
  this.vertx = vertx;
  hook = completionHandler -> {
    try {
      doClose();
      registries.remove(vertx);
    } catch (Exception e) {
      completionHandler.handle(Future.failedFuture(e));
      return;
    }
    completionHandler.handle(Future.succeededFuture());
  };
  vertx.addCloseHook(hook);
}

protected void kill(int pos) {
  VertxInternal v = (VertxInternal) vertices[pos];
  v.executeBlocking(fut -> {
    v.simulateKill();
    fut.complete();
  }, ar -> {
    assertTrue(ar.succeeded());
  });
}

public JDBCClientImpl(Vertx vertx, DataSource dataSource) {
  Objects.requireNonNull(vertx);
  Objects.requireNonNull(dataSource);
  this.vertx = vertx;
  this.holder = new DataSourceHolder((VertxInternal) vertx, dataSource);
  this.exec = holder.exec();
  this.ds = dataSource;
  this.metrics = holder.metrics;
  this.helper = new JDBCStatementHelper();
  setupCloseHook();
}

private DataSourceHolder lookupHolder(String datasourceName, JsonObject config) {
  synchronized (vertx) {
    LocalMap<String, DataSourceHolder> map = vertx.sharedData().getLocalMap(DS_LOCAL_MAP_NAME);
    DataSourceHolder theHolder = map.get(datasourceName);
    if (theHolder == null) {
      theHolder = new DataSourceHolder((VertxInternal) vertx, config, map, datasourceName);
    } else {
      theHolder.incRefCount();
    }
    return theHolder;
  }
}

DataSourceHolder(VertxInternal vertx, DataSource ds) {
  this.ds = ds;
  this.metrics = vertx.metricsSPI() != null ? vertx.metricsSPI().createMetrics(ds, "datasource", UUID.randomUUID().toString(), -1) : null;
  this.vertx = vertx;
  this.map = null;
  this.name = null;
}

DataSourceHolder(VertxInternal vertx, JsonObject config, LocalMap<String, DataSourceHolder> map, String name) {
  this.config = config;
  this.map = map;
  this.vertx = vertx;
  this.name = name;

  map.put(name, this);
}

@Override
public void start() throws Exception {

  vertx.runOnContext(v -> {
    List<String> extraCP = Arrays.asList("blah", "wibble");
    DeploymentOptions expected = new DeploymentOptions().setConfig(new JsonObject().put("foo", "bar"))
      .setWorker(true).setIsolationGroup("mygroup").setExtraClasspath(extraCP);
    Deployment dep = ((VertxInternal) vertx).getDeployment(Vertx.currentContext().deploymentID());
    vertx.eventBus().publish("moduleStarted", expected.equals(dep.deploymentOptions()));
  });
}

@Override
public void render(RoutingContext context, String templateDirectory, String templateFileName, Handler<AsyncResult<Buffer>> handler) {
  try {
    templateFileName = templateDirectory + templateFileName;
    CompiledTemplate template = isCachingEnabled() ? cache.get(templateFileName) : null;
    if (template == null) {
      // real compile
      String loc = adjustLocation(templateFileName);
      String templateText = Utils.readFileToString(context.vertx(), loc);
      if (templateText == null) {
        throw new IllegalArgumentException("Cannot find template " + loc);
      }
      template = TemplateCompiler.compileTemplate(templateText);
      if (isCachingEnabled()) {
        cache.put(templateFileName, template);
      }
    }
    Map<String, RoutingContext> variables = new HashMap<>(1);
    variables.put("context", context);
    final VertxInternal vertxInternal = (VertxInternal) context.vertx();
    String directoryName = vertxInternal.resolveFile(templateFileName).getParent();
    handler.handle(Future.succeededFuture(
      Buffer.buffer(
        (String) new TemplateRuntime(template.getTemplate(), null, template.getRoot(), directoryName)
          .execute(new StringAppender(), variables, new ImmutableDefaultFactory())
      )
    ));
  } catch (Exception ex) {
    handler.handle(Future.failedFuture(ex));
  }
}

@Override
public void setUp() throws Exception {
  super.setUp();
  // Setup our shiro+stormpath+vertx integration
  File file = ((VertxInternal) vertx).resolveFile("stormpath.properties");
  ApiKey apiKey = ApiKeys.builder().setFileLocation(file.getAbsolutePath()).build();
  Client client = Clients.builder().setApiKey(apiKey).build();
  ApplicationRealm stormpathAppRealm = new ApplicationRealm();
  stormpathAppRealm.setClient(client);
  stormpathAppRealm.setApplicationRestUrl("https://api.stormpath.com/v1/applications/2oFtzixwgN0wYKt25euKpg");
  authProvider = ShiroAuth.create(vertx, stormpathAppRealm);
}

@Override
public void setVertx(final Vertx vertx) {
    this.vertx = (VertxInternal) vertx;
}

public void registerCloseHook(VertxInternal vertx) {
  registerCloseHook(vertx::addCloseHook, vertx::removeCloseHook);
}

public static CommandRegistry get(Vertx vertx) {
  return registries.computeIfAbsent(vertx, v -> new CommandRegistryImpl((VertxInternal) vertx));
}

/**
 * Create a RedisConnection.
 */
public RedisConnection(Vertx vertx, RedisOptions config, RedisSubscriptions subscriptions) {

  // Make sure we have an event loop context for serializability of the commands
  Context ctx = Vertx.currentContext();
  if (ctx == null) {
    ctx = vertx.getOrCreateContext();
  } else if (!ctx.isEventLoopContext()) {
    VertxInternal vi = (VertxInternal) vertx;
    ctx = vi.createEventLoopContext(null, null, new JsonObject(), Thread.currentThread().getContextClassLoader());
  }

  this.vertx = vertx;
  this.context = ctx;
  this.config = config;

  this.subscriptions = subscriptions;

  if (subscriptions != null) {
    this.replyParser = new ReplyParser(reply -> {
      // Pub/sub messages are always multi-bulk
      if (reply.is('*')) {
        Reply[] data = (Reply[]) reply.data();
        if (data != null) {
          // message
          if (data.length == 3) {
            if (data[0].is('$') && "message".equals(data[0].asType(String.class))) {
              String channel = data[1].asType(String.class);
              subscriptions.handleChannel(channel, data);
              return;
            }
          }
          // pmessage
          else if (data.length == 4) {
            if (data[0].is('$') && "pmessage".equals(data[0].asType(String.class))) {
              String pattern = data[1].asType(String.class);
              subscriptions.handlePattern(pattern, data);
              return;
            }
          }
        }
      }

      // fallback to normal handler
      handleReply(reply);
    });

  } else {
    this.replyParser = new ReplyParser(this::handleReply);
  }
}

public HazelcastAsyncMultiMap(Vertx vertx, com.hazelcast.core.MultiMap<K, V> map) {
  this.vertx = (VertxInternal) vertx;
  this.map = map;
  map.addEntryListener(this, true);
}

/**
 * Create a new registry.
 *
 * @param vertx the vertx instance
 * @return the created registry
 */
static CommandRegistry create(Vertx vertx) {
  return new CommandRegistryImpl((VertxInternal) vertx);
}