Java 类io.vertx.core.net.NetServerOptions 实例源码

@SuppressWarnings({"rawtypes", "unchecked"})
@Test
public void testTcpServerSSL(@Mocked Vertx vertx, @Mocked AsyncResultCallback<InetSocketAddress> callback,
    @Mocked NetServer netServer) {
  new Expectations() {
    {
      vertx.createNetServer((NetServerOptions) any);
      result = netServer;
      netServer.connectHandler((Handler) any);
      netServer.listen(anyInt, anyString, (Handler) any);
    }
  };
  URIEndpointObject endpointObject = new URIEndpointObject("highway://127.0.0.1:6663?sslEnabled=true");
  TcpServer server = new TcpServerForTest(endpointObject);
  // assert done in Expectations
  server.init(vertx, "", callback);
}

@Override
public void start() throws Exception {
    NetServerOptions options = new NetServerOptions().setPort(8380);

    NetServer server = vertx.createNetServer(options);
    server.connectHandler(this::handle);
    server.close(result -> {
        if(result.succeeded()){
            //TCP server fully closed
            System.out.println("server close succeeded.");
        }else {
            System.out.println("server status: " + result.result().toString());
        }
    });
    server.listen();
}

private void startTcpServer(ConfigParser c) {
    int port = c.getPort();
    String keyPath = c.getTlsKeyPath();
    String certPath = c.getTlsCertPath();
    boolean tlsEnabled = c.isTlsEnabled();
    int idleTimeout = c.getSocketIdleTimeout();

    // MQTT over TCP
    NetServerOptions opt = new NetServerOptions()
            .setTcpKeepAlive(true)
            .setIdleTimeout(idleTimeout) // in seconds; 0 means "don't timeout".
            .setPort(port);

    if(tlsEnabled) {
        opt.setSsl(true).setPemKeyCertOptions(new PemKeyCertOptions()
            .setKeyPath(keyPath)
            .setCertPath(certPath)
        );
    }
    NetServer netServer = vertx.createNetServer(opt);
    Map<String, MQTTSession> sessions = new MonitoredMap<>();
    netServer.connectHandler(netSocket -> {
        MQTTNetSocket mqttNetSocket = new MQTTNetSocket(vertx, c, netSocket, sessions);
        mqttNetSocket.start();
    }).listen();
}

public void init(Vertx vertx, String sslKey, AsyncResultCallback<InetSocketAddress> callback) {
  NetServer netServer;
  if (endpointObject.isSslEnabled()) {
    SSLOptionFactory factory =
        SSLOptionFactory.createSSLOptionFactory(sslKey, null);
    SSLOption sslOption;
    if (factory == null) {
      sslOption = SSLOption.buildFromYaml(sslKey);
    } else {
      sslOption = factory.createSSLOption();
    }
    SSLCustom sslCustom = SSLCustom.createSSLCustom(sslOption.getSslCustomClass());
    NetServerOptions serverOptions = new NetServerOptions();
    VertxTLSBuilder.buildNetServerOptions(sslOption, sslCustom, serverOptions);
    netServer = vertx.createNetServer(serverOptions);
  } else {
    netServer = vertx.createNetServer();
  }

  netServer.connectHandler(netSocket -> {
    TcpServerConnection connection = createTcpServerConnection();
    connection.init(netSocket);
  });

  InetSocketAddress socketAddress = endpointObject.getSocketAddress();
  netServer.listen(socketAddress.getPort(), socketAddress.getHostString(), ar -> {
    if (ar.succeeded()) {
      callback.success(socketAddress);
      return;
    }

    // 监听失败
    String msg = String.format("listen failed, address=%s", socketAddress.toString());
    callback.fail(new Exception(msg, ar.cause()));
  });
}

public static NetServerOptions buildNetServerOptions(SSLOption sslOption, SSLCustom sslCustom,
    NetServerOptions netServerOptions) {
  buildTCPSSLOptions(sslOption, sslCustom, netServerOptions);
  if (sslOption.isAuthPeer()) {
    netServerOptions.setClientAuth(ClientAuth.REQUIRED);
  } else {
    netServerOptions.setClientAuth(ClientAuth.REQUEST);
  }
  return netServerOptions;
}

/**
 * Copies TLS trust store configuration to a given set of server options.
 * <p>
 * The trust store configuration is taken from <em>config</em> and will
 * be added only if the <em>ssl</em> flag is set on the given server options.
 * 
 * @param serverOptions The options to add configuration to.
 */
protected final void addTlsTrustOptions(final NetServerOptions serverOptions) {

    if (serverOptions.isSsl() && serverOptions.getTrustOptions() == null) {

        TrustOptions trustOptions = getConfig().getTrustOptions();
        if (trustOptions != null) {
            serverOptions.setTrustOptions(trustOptions).setClientAuth(ClientAuth.REQUEST);
            LOG.info("enabling TLS for client authentication");
        }
    }
}

/**
 * Copies TLS key & certificate configuration to a given set of server options.
 * <p>
 * If <em>config</em> contains key &amp; certificate configuration it is added to
 * the given server options and the <em>ssl</em> flag is set to {@code true}.
 * 
 * @param serverOptions The options to add configuration to.
 */
protected final void addTlsKeyCertOptions(final NetServerOptions serverOptions) {

    KeyCertOptions keyCertOptions = getConfig().getKeyCertOptions();

    if (keyCertOptions != null) {
        serverOptions.setSsl(true).setKeyCertOptions(keyCertOptions);
    }
}

public TcpEventBusBridgeImpl(Vertx vertx, BridgeOptions options, NetServerOptions netServerOptions, Handler<BridgeEvent> eventHandler) {
  this.eb = vertx.eventBus();
  this.options = options != null ? options : new BridgeOptions();
  this.bridgeEventHandler = eventHandler;

  server = vertx.createNetServer(netServerOptions == null ? new NetServerOptions() : netServerOptions);
  server.connectHandler(this::handler);
}

/**
 * Starts the server to listen on the configured port.
 */
public void start()
{
    if (vertx == null)
    {
        vertx = Vertx.vertx();
    }

    NetServerOptions options = new NetServerOptions();
    options.setTcpKeepAlive(true);
    options.setTcpNoDelay(true);

    srvVertx = vertx.createNetServer(options);

    srvVertx.connectHandler(new Handler<NetSocket>()
    {
        public void handle(NetSocket pSocket)
        {
            AbstractDataHandler dataHandler = new NetDataHandler(srvJVx, pSocket); 

            pSocket.handler(dataHandler);
            pSocket.endHandler(new StopHandler(dataHandler));
            pSocket.exceptionHandler(new ExceptionHandler(dataHandler));
        }
    });

    srvVertx.listen(iPort, sInterface);
}

protected NetServer createNetServer(NetServerOptions options) {
  NetServer server = vertx.createNetServer(options);
  toClose.add(() -> {
    CountDownLatch latch = new CountDownLatch(1);
    server.close(ar -> {
      latch.countDown();
    });
    awaitLatch(latch);
    return null;
  });
  return server;
}

@Test
public void testNetMetricsOnClose() throws Exception {
  int requests = 8;
  CountDownLatch latch = new CountDownLatch(requests);

  NetClient client = vertx.createNetClient(new NetClientOptions());
  NetServer server = vertx.createNetServer(new NetServerOptions().setHost("localhost").setPort(1235).setReceiveBufferSize(50)).connectHandler(socket -> {
    socket.handler(buff -> latch.countDown());
  }).listen(ar -> {
    assertTrue(ar.succeeded());
    client.connect(1235, "localhost", ar2 -> {
      assertTrue(ar2.succeeded());
      for (int i = 0; i < requests; i++) {
        ar2.result().write(randomBuffer(50));
      }
    });
  });

  awaitLatch(latch);

  client.close();
  server.close(ar -> {
    assertTrue(ar.succeeded());
    vertx.runOnContext(v -> testComplete());
  });

  await();

  JsonObject metrics = metricsService.getMetricsSnapshot(server);
  assertNotNull(metrics);
  assertTrue(metrics.isEmpty());

  metrics = metricsService.getMetricsSnapshot(client);
  assertNotNull(metrics);
  assertTrue(metrics.isEmpty());

  cleanup(client);
  cleanup(server);
}

@Test
public void testMetricsCleanupedOnVertxClose() throws Exception {
  CountDownLatch latch1 = new CountDownLatch(1);
  HttpServer server = vertx.createHttpServer(new HttpServerOptions().setPort(8080));
  server.requestHandler(req -> {});
  server.listen(onSuccess(res -> {
    latch1.countDown();
  }));
  awaitLatch(latch1);
  HttpClient client = vertx.createHttpClient(new HttpClientOptions());
  CountDownLatch latch2 = new CountDownLatch(1);
  NetServer nServer = vertx.createNetServer(new NetServerOptions().setPort(1234));
  nServer.connectHandler(conn -> {});
  nServer.listen(res -> {
    latch2.countDown();
  });
  awaitLatch(latch2);
  NetClient nClient = vertx.createNetClient(new NetClientOptions());
  DatagramSocket sock = vertx.createDatagramSocket(new DatagramSocketOptions());
  EventBus eb = vertx.eventBus();
  assertFalse(metricsService.getMetricsSnapshot(vertx).isEmpty());
  assertFalse(metricsService.getMetricsSnapshot(server).isEmpty());
  assertFalse(metricsService.getMetricsSnapshot(client).isEmpty());
  assertFalse(metricsService.getMetricsSnapshot(nServer).isEmpty());
  assertFalse(metricsService.getMetricsSnapshot(nClient).isEmpty());
  assertFalse(metricsService.getMetricsSnapshot(sock).isEmpty());
  assertFalse(metricsService.getMetricsSnapshot(eb).isEmpty());
  vertx.close(res -> {
    assertTrue(metricsService.getMetricsSnapshot(vertx).isEmpty());
    assertTrue(metricsService.getMetricsSnapshot(server).isEmpty());
    assertTrue(metricsService.getMetricsSnapshot(client).isEmpty());
    assertTrue(metricsService.getMetricsSnapshot(nServer).isEmpty());
    assertTrue(metricsService.getMetricsSnapshot(nClient).isEmpty());
    assertTrue(metricsService.getMetricsSnapshot(sock).isEmpty());
    assertTrue(metricsService.getMetricsSnapshot(eb).isEmpty());
    testComplete();
  });
  await();
  vertx = null;
}

public void start() {
    //TODO: Fix a better way of configuration other than system properties?
    Integer port = Integer.getInteger("tcp.port", 5555);

    ObservableFuture<NetServer> netServerObservable = RxHelper.observableFuture();
    NetServer netServer = vertx.createNetServer(new NetServerOptions().setPort(port));
    netServerObservable.subscribe(a ->
            log.info("Starting TCP listener.."),
            e -> log.error("Could not start TCP listener on port " + port, e),
            () -> log.info("Started TCP listener on port " + port + ".")
    );

    RxHelper.toObservable(netServer.connectStream())
            .flatMap(s -> Riemann.convertBufferStreamToMessages(s, RxHelper.toObservable(s)))
            .subscribe(s -> {
                sendResponse(Proto.Msg.newBuilder().setOk(true).build(), s.getLeft());
                vertx.eventBus().publish("riemann.stream", s.getRight().toByteArray());
            }, e -> {
                log.error(e);

                if (e instanceof NetSocketException) {
                    sendResponse(Proto.Msg.newBuilder().setError(e.getMessage()).build(), ((NetSocketException) e).getSocket());
                }
            });

    netServer.listen(netServerObservable.asHandler());
}

private NetServer startTcpBroker(JsonObject conf) {

        ConfigParser c = new ConfigParser(conf);

        NetServerOptions opt = new NetServerOptions().setTcpKeepAlive(true)
                .setIdleTimeout(conf.getInteger("socket_idle_timeout")).setPort(conf.getInteger("tcp_port"));

        NetServer netServer = vertx.createNetServer(opt);
        netServer.connectHandler(netSocket -> {
            Map<String, MQTTSession> sessions = new HashMap<>();
            MQTTNetSocket mqttNetSocket = new MQTTNetSocket(vertx, c, netSocket, sessions);
            mqttNetSocket.start();
        }).listen();
        return netServer;
    }

@Override
default TCPMetrics<?> createMetrics(SocketAddress localAddress, NetServerOptions options) {
    return DummyVertxMetrics.DummyTCPMetrics.INSTANCE;
}

@Override
public TCPMetrics<?> createMetrics(SocketAddress localAddress, NetServerOptions options) {
    return this.<TCPMetrics<?>> createSubMetrics(v -> v.createMetrics(localAddress, options),
            d -> new DispatchingTcpMetrics(d));
}

@Override
public TCPMetrics<?> createMetrics(SocketAddress localAddress, NetServerOptions options) {
    return new NetServerMetricsImpl(counterService, gaugeService, properties.getNet().getServer());
}

@Override
public @NotNull TCPMetrics<?> createMetrics(@NotNull SocketAddress localAddress, @NotNull NetServerOptions netServerOptions) {
  return options.isEnabled(NetServer)
      ? new NetServerPrometheusMetrics(options.getRegistry(), localAddress)
      : super.createMetrics(localAddress, netServerOptions);
}

@Override
public NetServer createNetServer(NetServerOptions options) {
    return vertx.createNetServer(options);
}

static TcpEventBusBridge create(Vertx vertx, BridgeOptions options, NetServerOptions netServerOptions) {
  return new TcpEventBusBridgeImpl(vertx, options, netServerOptions,null);
}

static TcpEventBusBridge create(Vertx vertx, BridgeOptions options, NetServerOptions netServerOptions,Handler<BridgeEvent> eventHandler) {
  return new TcpEventBusBridgeImpl(vertx, options, netServerOptions,eventHandler);
}

public TcpEventBusBridgeImpl(Vertx vertx, BridgeOptions options, NetServerOptions netServerOptions) {
  this(vertx, options, netServerOptions, null);
}

@SuppressWarnings("serial")
public LinkManager(Vertx vertx, AmqpServiceConfig config, AMQPServiceImpl parent) {
  DEFAULT_OUTGOING_LINK_OPTIONS.setReliability(ReliabilityMode.AT_LEAST_ONCE);
  _vertx = vertx;
  _config = config;
  _listener = parent;
  _client = _vertx.createNetClient(new NetClientOptions());

  URL_CACHE = Collections.synchronizedMap(new LinkedHashMap<String, ConnectionSettings>(config
    .getMaxedCachedURLEntries() + 1, 1.1f, true) {
    @Override
    protected boolean removeEldestEntry(Map.Entry<String, ConnectionSettings> eldest) {
      return size() > config.getMaxedCachedURLEntries();
    }
  });

  NetServerOptions serverOp = new NetServerOptions();
  serverOp.setHost(config.getInboundHost());
  serverOp.setPort(config.getInboundPort());
  _server = _vertx.createNetServer(serverOp);
  _server.connectHandler(sock -> {
    DefaultConnectionSettings settings = new DefaultConnectionSettings();
    settings.setHost(sock.remoteAddress().host());
    settings.setPort(sock.remoteAddress().port());
    ManagedConnection connection = new ManagedConnection(settings, this, true);
    connection.setNetSocket(sock);
    connection.write();
    _inboundConnections.add(connection);
    connection.addDisconnectHandler(c -> {
      _inboundConnections.remove(c);
    });
  });

  _server.listen(result -> {
    if (result.failed()) {
      String error = format("Error {%s} Server was unable to bind to %s:%s", result.cause(),
        _config.getInboundHost(), _config.getInboundPort());
      LOG.fatal(error, result.cause());
      // We need to stop the verticle
      LOG.fatal("Initiating the shutdown of AMQP Service due to : %s", error);
      parent.stopInternal();
    }
  });
}

@Override
public TCPMetrics createMetrics(SocketAddress localAddress, NetServerOptions options) {
  NetServerMetricsSupplier supplier = (NetServerMetricsSupplier) metricSuppliers.get(NET_SERVER);
  return supplier != null ? new NetServerMetricsImpl(localAddress, supplier) : super.createMetrics(localAddress, options);
}

@Override
public TCPMetrics<?> createMetrics(SocketAddress localAddress, NetServerOptions options) {
  String baseName = MetricRegistry.name(nameOf("net.servers"), TCPMetricsImpl.addressName(localAddress));
  return new TCPMetricsImpl(registry, baseName);
}

public void start(final Future<Void> startedResult) {

        final NetServer netServer = vertx.createNetServer(new NetServerOptions().setAcceptBacklog(10000));
        logger.info("Echo is Hello world!");
        netServer
                .connectHandler(socket -> {
                    connectionCount.incrementAndGet();
                    socket.exceptionHandler(event -> {
                        logger.error("Socket error on echo service socket", event);
                    });

                    socket.closeHandler(v -> {
                        connectionCount.decrementAndGet();
                    });



                    Pump.pump(socket, socket).start();


                })
                .listen(ECHO_SERVICE_PORT, ECHO_SERVICE_HOST, event -> {
                    if (event.failed()) {
                        final Throwable cause = event.cause();
                        logger.error(cause.getMessage(), cause);
                        startedResult.fail(cause);
                        return;
                    }
                    logger.info(String.format("Started echo server - %s", ECHO_SERVICE_PORT));
                    startedResult.complete();
                });


    }

private NetServerOptions buildOptions(JsonObject configObj) {
    return new NetServerOptions(configObj);
}

@Override
public void start() throws Exception {
    address = MQTTSession.ADDRESS;

    JsonObject conf = config();

    localBridgePort = conf.getInteger("local_bridge_port", 7007);
    idleTimeout = conf.getInteger("socket_idle_timeout", 120);
    ssl_cert_key = conf.getString("ssl_cert_key");
    ssl_cert = conf.getString("ssl_cert");
    ssl_trust = conf.getString("ssl_trust");


    // [TCP -> BUS] listen TCP publish to BUS
    NetServerOptions opt = new NetServerOptions()
            .setTcpKeepAlive(true)
            .setIdleTimeout(idleTimeout)
            .setPort(localBridgePort)
    ;

    if(ssl_cert_key != null && ssl_cert != null && ssl_trust != null) {
        opt.setSsl(true).setClientAuth(ClientAuth.REQUIRED)
            .setPemKeyCertOptions(new PemKeyCertOptions()
                .setKeyPath(ssl_cert_key)
                .setCertPath(ssl_cert)
            )
            .setPemTrustOptions(new PemTrustOptions()
                .addCertPath(ssl_trust)
            )
        ;
    }

    netServer = vertx.createNetServer(opt);
    netServer.connectHandler(sock -> {
        final EventBusNetBridge ebnb = new EventBusNetBridge(sock, vertx.eventBus(), address);
        sock.closeHandler(aVoid -> {
            logger.info("Bridge Server - closed connection from client ip: " + sock.remoteAddress());
            ebnb.stop();
        });
        sock.exceptionHandler(throwable -> {
            logger.error("Bridge Server - Exception: " + throwable.getMessage(), throwable);
            ebnb.stop();
        });

        logger.info("Bridge Server - new connection from client ip: " + sock.remoteAddress());

        RecordParser parser = ebnb.initialHandhakeProtocolParser();
        sock.handler(parser::handle);

    }).listen();
}