com.codahale.metrics.RatioGauge#io.vertx.core.net.SocketAddress源码实例Demo

@Test
public void testHttpClientLegacy4(TestContext context) {
  Async async = context.async();
  StringBuilder b = new StringBuilder();

  context.assertTrue(server != null);
  HttpClient client = vertx.createHttpClient();
  SocketAddress sa = SocketAddress.inetSocketAddress(PORT + 1, LOCALHOST);
  HttpClientRequest requestAbs = HttpClientLegacy.requestAbs(client,
    HttpMethod.GET, sa, URL + "/test1", res -> {
      b.append("response");
      async.complete();
    });
  requestAbs.exceptionHandler(res -> {
    b.append("exception");
    async.complete();
  });
  requestAbs.end();
  async.await();
  context.assertEquals("exception", b.toString());
}
 

@Override
public synchronized ClientMetrics<?, ?, ?, ?> createClientMetrics(SocketAddress remoteAddress, String type, String namespace) {
  String baseName;
  if (namespace != null && namespace.length() > 0) {
    baseName = MetricRegistry.name(nameOf(type, "clients", namespace, remoteAddress.toString()));
  } else {
    baseName = MetricRegistry.name(nameOf(type, "clients", remoteAddress.toString()));
  }
  return clientMetrics.compute(baseName, (key, prev) -> {
    if (prev == null) {
      return new DropwizardClientMetrics<>(this, registry, baseName, 1);
    } else {
      return prev.inc();
    }
  });
}
 

@Override
public Long connected(SocketAddress remoteAddress, String remoteName) {
  // Connection metrics
  openConnections.inc();

  // On network outage the remoteAddress can be null.
  // Do not report the open-connections when it's null
  if (remoteAddress != null) {
    // Remote address connection metrics
    counter("open-connections", remoteAddress.host()).inc();

  }

  // A little clunky, but it's possible we got here after closed has been called
  if (closed) {
    removeAll();
  }

  return System.nanoTime();
}
 

@Test
public void testHttpClientLegacy3(TestContext context) {
  Async async = context.async();
  StringBuilder b = new StringBuilder();

  context.assertTrue(server != null);
  HttpClient client = vertx.createHttpClient();
  SocketAddress sa = SocketAddress.inetSocketAddress(PORT, LOCALHOST);
  HttpClientRequest requestAbs = HttpClientLegacy.requestAbs(client,
    HttpMethod.GET, sa, URL + "/test1", res -> {
      b.append("response");
      async.complete();
    });
  requestAbs.exceptionHandler(res -> {
    b.append("exception");
    async.complete();
  });
  requestAbs.end();
  async.await();
  context.assertEquals("response", b.toString());
}
 

HttpRequestImpl(WebClientInternal client, HttpMethod method, SocketAddress serverAddress, String protocol, Boolean ssl, Integer port, String host, String
        uri, BodyCodec<T> codec, WebClientOptions options) {
  this.client = client;
  this.method = method;
  this.protocol = protocol;
  this.codec = codec;
  this.port = port;
  this.host = host;
  this.uri = uri;
  this.ssl = ssl;
  this.serverAddress = serverAddress;
  this.followRedirects = options.isFollowRedirects();
  this.options = options;
  if (options.isUserAgentEnabled()) {
    headers = HttpHeaders.set(HttpHeaders.USER_AGENT, options.getUserAgent());
  }
}
 

public void testSocketAddress(WebClient client) {

    // Creates the unix domain socket address to access the Docker API
    SocketAddress serverAddress = SocketAddress
      .domainSocketAddress("/var/run/docker.sock");

    // We still need to specify host and port so the request
    // HTTP header will be localhost:8080
    // otherwise it will be a malformed HTTP request
    // the actual value does not matter much for this example
    client
      .request(
        HttpMethod.GET,
        serverAddress,
        8080,
        "localhost",
        "/images/json")
      .expect(ResponsePredicate.SC_ACCEPTED)
      .as(BodyCodec.jsonObject())
      .send()
      .onSuccess(res ->
        System.out.println("Current Docker images" + res.body()))
      .onFailure(err ->
        System.out.println("Something went wrong " + err.getMessage()));
  }
 

@Override
public InetSocketAddress getDestinationAddress() {
    SocketAddress socketAddress = request.localAddress();
    if (socketAddress == null) {
        return null;
    }
    return new InetSocketAddress(socketAddress.host(), socketAddress.port());
}
 

@Override
public InetSocketAddress getSourceAddress() {
    SocketAddress socketAddress = request.remoteAddress();
    if (socketAddress == null) {
        return null;
    }
    return new InetSocketAddress(socketAddress.host(), socketAddress.port());
}
 

@Override
public InetSocketAddress getRemoteAddress() {
    if (null == socket) {
        return null;
    }
    SocketAddress socketAddress = socket.remoteAddress();
    return new InetSocketAddress(socketAddress.host(), socketAddress.port());
}
 

@Override
public InetSocketAddress getRemoteAddress() {
    SocketAddress address = delegate.remoteAddress();
    if (address == null) {
        return null;
    }
    return new InetSocketAddress(address.host(), address.port());
}
 

DockerModuleHandle(Vertx vertx, LaunchDescriptor desc,
                   String id, Ports ports, String containerHost, int port, JsonObject config) {
  this.hostPort = port;
  this.ports = ports;
  this.id = id;
  this.containerHost = containerHost;
  this.image = desc.getDockerImage();
  this.cmd = desc.getDockerCmd();
  this.env = desc.getEnv();
  this.dockerArgs = desc.getDockerArgs();
  this.client = vertx.createHttpClient();
  this.logBuffer = new StringBuilder();
  this.logSkip = 0;
  logger.info("Docker handler with native: {}", vertx.isNativeTransportEnabled());
  Boolean b = desc.getDockerPull();
  this.dockerPull = b == null || b.booleanValue();
  StringBuilder socketFile = new StringBuilder();
  this.dockerUrl = setupDockerAddress(socketFile,
      Config.getSysConf("dockerUrl", DEFAULT_DOCKER_URL, config));
  if (socketFile.length() > 0) {
    socketAddress = SocketAddress.domainSocketAddress(socketFile.toString());
  } else {
    socketAddress = null;
  }
  tcpPortWaiting = new TcpPortWaiting(vertx, containerHost, port);
  if (desc.getWaitIterations() != null) {
    tcpPortWaiting.setMaxIterations(desc.getWaitIterations());
  }
}
 

private void checkDocker(Handler<AsyncResult<JsonArray>> future) {
  final SocketAddress socketAddress
    = SocketAddress.domainSocketAddress("/var/run/docker.sock");
  final String url = "http://localhost/images/json?all=1";
  HttpClientRequest req = HttpClientLegacy.requestAbs(client, HttpMethod.GET,
    socketAddress, url, res -> {
      Buffer body = Buffer.buffer();
      res.handler(d -> {
        body.appendBuffer(d);
      });
      res.endHandler(d -> {
        if (res.statusCode() == 200) {
          boolean gotIt = false;
          try {
            JsonArray ar = body.toJsonArray();
            future.handle(Future.succeededFuture(ar));
          } catch (Exception ex) {
            logger.warn(ex);
            future.handle(Future.failedFuture(ex));
          }
        } else {
          String m = "checkDocker HTTP error " + res.statusCode() + "\n"
            + body.toString();
          logger.error(m);
          future.handle(Future.failedFuture(m));
        }
      });
      res.exceptionHandler(d -> {
        logger.warn("exceptionHandler 2 " + d, d);
        future.handle(Future.failedFuture(d));
      });
    });
  req.exceptionHandler(d -> {
    logger.warn("exceptionHandler 1 " + d, d);
    future.handle(Future.failedFuture(d));
  });
  req.end();
}
 

/**
 * Send HTTP request with style ala Vert.x 3.
 * @param client HTTP client
 * @param method HTTP method
 * @param socketAddress socket address
 * @param url Full URL
 * @param response response handler
 * @return handler for HTTP request
 */
public static HttpClientRequest requestAbs(HttpClient client, HttpMethod method,
                                           SocketAddress socketAddress, String url,
                                           Handler<HttpClientResponse> response) {
  return client.requestAbs(method, socketAddress, url, hndlr -> {
    if (hndlr.succeeded()) {
      response.handle(hndlr.result());
    }
  });
}
 

@Override
public HttpServerMetrics<?, ?, ?> createHttpServerMetrics(HttpServerOptions httpClientOptions, SocketAddress socketAddress) {
  if (httpServerMetrics != null) {
    return httpServerMetrics.forAddress(socketAddress);
  }
  return DummyVertxMetrics.DummyHttpServerMetrics.INSTANCE;
}
 

@Override
public TCPMetrics<?> createNetServerMetrics(NetServerOptions netServerOptions, SocketAddress socketAddress) {
  if (netServerMetrics != null) {
    return netServerMetrics.forAddress(socketAddress);
  }
  return DummyVertxMetrics.DummyTCPMetrics.INSTANCE;
}
 

@Override
public String readAttribute(final RoutingContext exchange) {
    SocketAddress localAddress = exchange.request().localAddress();
    if (localAddress == null) {
        return null;
    }
    return localAddress.host();
}
 

private SocketAddress parseFor(String forToParse, int defaultPort) {
  String host = forToParse;
  int port = defaultPort;
  int portSeparatorIdx = forToParse.lastIndexOf(':');
  if (portSeparatorIdx > forToParse.lastIndexOf(']')) {
    host = forToParse.substring(0, portSeparatorIdx);
    port = parsePort(forToParse.substring(portSeparatorIdx + 1), defaultPort);
  }

  return new SocketAddressImpl(port, host);
}
 

@Override
public String readAttribute(final RoutingContext exchange) {
    final SocketAddress sourceAddress = exchange.request().remoteAddress();
    if (sourceAddress == null) {
        return null;
    }
    return sourceAddress.host();
}
 

private SocketAddress parseFor(String forToParse, int defaultPort) {
    String host = forToParse;
    int port = defaultPort;
    int portSeparatorIdx = forToParse.lastIndexOf(':');
    if (portSeparatorIdx > forToParse.lastIndexOf(']')) {
        host = forToParse.substring(0, portSeparatorIdx);
        port = parsePort(forToParse.substring(portSeparatorIdx + 1), defaultPort);
    }

    return new SocketAddressImpl(port, host);
}
 

private void setupUnixDomainSocketHttpServer(HttpServer httpServer, HttpServerOptions options, Future<Void> startFuture,
        AtomicInteger remainingCount) {
    httpServer.listen(SocketAddress.domainSocketAddress(options.getHost()), event -> {
        if (event.succeeded()) {
            if (remainingCount.decrementAndGet() == 0) {
                startFuture.complete(null);
            }
        } else {
            startFuture.fail(event.cause());
        }
    });
}
 

/**
 * 调用微信统一下单接口
 *
 * @param product 充值设备描述
 * @param price   充值设备价格
 * @param ip      充值端Ip
 * @param openId  充值的微信openId
 * @param acc     账户对象
 * 异步返回 微信统一下单接口返回的xml数据（String）
 *
 * @author Leibniz
 */
private void unifyPay(String orderId, String product, int price, SocketAddress ip, String openId, String attach, String notUrl, JsonObject acc, Handler<String> callback) {
    Map<String, Object> map = new TreeMap<>();
    map.put("appid", acc.getString(WXAPPID));
    map.put("mch_id", acc.getString(MCHID));
    map.put("nonce_str", CommonUtils.getRandomID());
    map.put("body", product);
    map.put("out_trade_no", orderId);
    map.put("total_fee", price);
    map.put("spbill_create_ip", ip.host());
    map.put("notify_url", notUrl);
    map.put("trade_type", "JSAPI");
    map.put("openid", openId);
    if (null != attach) {
        map.put("attach", attach);
    }
    map.put("sign", WechatPay.getWeixinPaySign(map, acc.getString(MCHKEY)));
    String xmlStr = null;
    String encode = "ISO8859-1";
    try {
        xmlStr = XmlUtils.simpleMapToXml(map, encode);
    } catch (UnsupportedEncodingException e) {
        e.printStackTrace();
    }
    log.debug("下单请求数据：" + xmlStr);
    NetworkUtils.asyncPostStringWithData(WECHAT_UNIFY_PAY, xmlStr, XML, encode, callback);
}
 

@Override
public DefaultTcpSocketMetric connected(SocketAddress remoteAddress, String remoteName) {
  DefaultClientEndpointMetric endpointMetric = this.clientEndpointMetricManager
      .getOrCreateEndpointMetric(remoteAddress.toString());
  endpointMetric.onConnect();
  return new DefaultTcpSocketMetric(endpointMetric);
}
 

@Before
public void setup() {
  vertx = Vertx.vertx(new VertxOptions().setMetricsOptions(
    new MetricsOptions().setEnabled(true).setFactory(tracingOptions -> new VertxMetrics() {
      @Override
      public ClientMetrics<?, ?, ?, ?> createClientMetrics(SocketAddress remoteAddress, String type, String namespace) {
        return metrics;
      }
    }))
  );
}
 

@Test
public void testGetLocalPort(@Mocked SocketAddress sa) {
  new Expectations() {
    {
      sa.port();
      result = 1234;
      vertxRequest.localAddress();
      result = sa;
    }
  };

  Assert.assertEquals(1234, request.getLocalPort());
}
 

private String getLocalAddress() {
  HttpConnection connection = clientRequest.connection();
  if (connection == null) {
    return "not connected";
  }
  SocketAddress socketAddress = connection.localAddress();
  return socketAddress != null ? socketAddress.toString() : "not connected";
}
 

@Before
public void initStrBuilder() {
  routingContext = mock(RoutingContext.class);
  finishEvent = mock(InvocationFinishEvent.class);
  invocation = mock(Invocation.class);
  serverRequest = mock(HttpServerRequest.class);
  endpoint = mock(Endpoint.class);
  uriEndpointObject = mock(URIEndpointObject.class);
  socketAddress = mock(SocketAddress.class);
  accessLogEvent = new ServerAccessLogEvent();
  accessLogEvent.setRoutingContext(routingContext);
  strBuilder = new StringBuilder();
}
 

private Handler<ServerWebSocket> websocketHandler() {
  return websocket -> {
    final SocketAddress socketAddress = websocket.remoteAddress();
    final String connectionId = websocket.textHandlerID();
    final String token = getAuthToken(websocket);
    if (token != null) {
      LOG.trace("Websocket authentication token {}", token);
    }

    if (!hasAllowedHostnameHeader(Optional.ofNullable(websocket.headers().get("Host")))) {
      websocket.reject(403);
    }

    LOG.debug("Websocket Connected ({})", socketAddressAsString(socketAddress));

    websocket.textMessageHandler(
        payload -> {
          LOG.debug(
              "Received Websocket request {} ({})",
              payload,
              socketAddressAsString(socketAddress));

          AuthenticationUtils.getUser(
              authenticationService,
              token,
              user ->
                  websocketRequestHandler.handle(
                      authenticationService, connectionId, payload, user));
        });

    websocket.closeHandler(
        v -> {
          LOG.debug("Websocket Disconnected ({})", socketAddressAsString(socketAddress));
          vertx
              .eventBus()
              .publish(SubscriptionManager.EVENTBUS_REMOVE_SUBSCRIPTIONS_ADDRESS, connectionId);
        });

    websocket.exceptionHandler(
        t -> {
          LOG.debug(
              "Unrecoverable error on Websocket: {} ({})",
              t.getMessage(),
              socketAddressAsString(socketAddress));
          websocket.close();
        });
  };
}
 

private SocketAddress resolveAddress() {
    String host = resourceReader.readProperty(ServerProperty.SERVER_HOST, String.class, ServerProperty.DEFAULT_HOST);
    Integer port = resourceReader.readProperty(ServerProperty.SERVER_PORT, Integer.class, ServerProperty.DEFAULT_PORT);
    return SocketAddress.inetSocketAddress(port, host);
}
 

@Override
public synchronized SocketAddress localAddress() {
  return localAddress;
}
 

@Override
public String localAddress() {
    SocketAddress address = httpServerRequest.localAddress();
    return (address != null) ? address.host() : null;
}