/**
 * Opens an {@link HttpURLConnection} with parameters.
 * @param url
 * @return an open connection
 * @throws IOException
 */
private HttpURLConnection openConnection(URL url, Request<?> request) throws IOException {
    HttpURLConnection connection = createConnection(url);

    int timeoutMs = request.getTimeoutMs();
    connection.setConnectTimeout(timeoutMs);
    connection.setReadTimeout(timeoutMs);
    connection.setUseCaches(false);
    connection.setDoInput(true);

    // use caller-provided custom SslSocketFactory, if any, for HTTPS
    if ("https".equals(url.getProtocol()) && mSslSocketFactory != null) {
        ((HttpsURLConnection)connection).setSSLSocketFactory(mSslSocketFactory);
    }

    return connection;
}
 

/**
 * This method returns the entire result from the HTTP response.
 *
 * @param url The URL to fetch the HTTP response from.
 * @return The contents of the HTTP response.
 * @throws IOException Related to network and stream reading
 */
public static String getResponseFromHttpUrl(URL url) throws IOException {
    HttpURLConnection urlConnection = (HttpURLConnection) url.openConnection();
    try {
        InputStream in = urlConnection.getInputStream();

        Scanner scanner = new Scanner(in);
        scanner.useDelimiter("\\A");

        boolean hasInput = scanner.hasNext();
        if (hasInput) {
            return scanner.next();
        } else {
            return null;
        }
    } finally {
        urlConnection.disconnect();
    }
}
 

@RunAsClient
@Test
public void testSessionSwitched(@ArquillianResource URL baseURL) throws IOException {
  URL urlTest = url(baseURL, "TestServlet", "testSessionSwitched");
  String originalSession = callWebapp(urlTest);
  HttpURLConnection connection = (HttpURLConnection) url(baseURL, "TestServlet" + originalSession,
                                                         "testSessionSwitched").openConnection();
  assertThat(connection, matchLines("Previous value of attribute: B", "New value of attribute: B", "Encoded url: /" + originalSession));

  URL urlOther = url(baseURL, "SwitchServlet" + originalSession, "testSessionSwitched");
  String switchedSession = callWebapp(urlOther);

  assertThat(switchedSession, not(equalTo(originalSession)));
  HttpURLConnection connectionOther = (HttpURLConnection) url(baseURL, "TestServlet" + switchedSession,
                                                              "testSessionSwitched").openConnection();
  assertThat(connectionOther, matchLines("Previous value of attribute: S", "New value of attribute: B"));
}
 

@Override
public <T extends BoxObject> T onResponse(Class<T> clazz, BoxHttpResponse response) throws IllegalAccessException, InstantiationException, BoxException {
        if (response.getResponseCode() == HttpURLConnection.HTTP_ACCEPTED) {
            try {
                // First attempt to use Retry-After header, all failures will eventually fall back to exponential backoff
                if (mNumAcceptedRetries < DEFAULT_NUM_RETRIES) {
                    mNumAcceptedRetries++;
                    mRetryAfterMillis = getRetryAfterFromResponse(response, 1);
                } else if (mRetryAfterMillis < DEFAULT_MAX_WAIT_MILLIS) {
                    // Exponential back off with some randomness to avoid traffic spikes to server
                    mRetryAfterMillis *= (1.5 + Math.random());
                } else {
                    // Give up after the maximum retry time is exceeded.
                    throw new BoxException.MaxAttemptsExceeded("Max wait time exceeded.", mNumAcceptedRetries);
                }
                Thread.sleep(mRetryAfterMillis);
                return (T) mRequest.send();
            } catch (InterruptedException e) {
                throw new BoxException(e.getMessage(), response);
            }
        } else {
            BoxIterator list = super.onResponse(BoxIteratorBoxEntity.class, response);
            return (T)list.get(0);
        }
    }
 

private HttpResponse doFetch() throws IOException {
  HttpURLConnection conn = configureAndConnectRequest();
  try {
    ByteArrayOutputStream out = new ByteArrayOutputStream();
    InputStream rawStream = conn.getInputStream();
    try {
      // Let Stetho see the raw, possibly compressed stream.
      rawStream = stethoManager.interpretResponseStream(rawStream);
      InputStream decompressedStream = applyDecompressionIfApplicable(conn, rawStream);
      if (decompressedStream != null) {
        copy(decompressedStream, out, new byte[1024]);
      }
    } finally {
      if (rawStream != null) {
        rawStream.close();
      }
    }
    return new HttpResponse(conn.getResponseCode(), out.toByteArray());
  } finally {
    conn.disconnect();
  }
}
 

/**
 * Tests updating an OFAgent with PUT.
 */
@Test
public void testOFAgentUpdate() {
    expect(mockOFAgentService.agent(eq(NETWORK))).andReturn(OF_AGENT).anyTimes();
    replay(mockOFAgentService);

    mockOFAgentAdminService.updateAgent(anyObject());
    expectLastCall().anyTimes();
    replay(mockOFAgentAdminService);

    InputStream jsonStream = OFAgentWebResourceTest.class
            .getResourceAsStream("put-ofagent-update.json");
    assertNotNull("put-ofagent-update.json is null", jsonStream);
    WebTarget wt = target();
    Response response = wt.path("service/ofagent-update")
            .request(MediaType.APPLICATION_JSON_TYPE)
            .put(Entity.json(jsonStream));
    assertThat(response.getStatus(), is(HttpURLConnection.HTTP_OK));
    assertThat(response.readEntity(String.class), containsString("OFAgent updated"));

    verify(mockOFAgentService);
    verify(mockOFAgentAdminService);

}
 

public List<DockerContainer> getContainers() {
    try {
        JSONArray value = (JSONArray) doGetRequest("/containers/json?all=1",
                Collections.singleton(HttpURLConnection.HTTP_OK));
        List<DockerContainer> ret = new ArrayList<>(value.size());
        for (Object o : value) {
            JSONObject json = (JSONObject) o;
            String id = (String) json.get("Id");
            String image = (String) json.get("Image");
            String name = null;
            JSONArray names = (JSONArray) json.get("Names");
            if (names != null && !names.isEmpty()) {
                name = (String) names.get(0);
            }
            DockerContainer.Status status = DockerUtils.getContainerStatus((String) json.get("Status"));
            ret.add(new DockerContainer(instance, id, image, name, status));
        }
        return ret;
    } catch (DockerException ex) {
        LOGGER.log(Level.INFO, null, ex);
    }
    return Collections.emptyList();
}
 

private ConsentInfoUpdateResponse makeConsentLookupRequest(String urlString) {
    try {
        URL url = new URL(urlString);

        HttpURLConnection urlConnection = (HttpURLConnection) url.openConnection();
        if (urlConnection.getResponseCode() == HttpURLConnection.HTTP_OK) {
            String responseString = readStream(urlConnection.getInputStream());
            urlConnection.disconnect();
            consentInformation.updateConsentData(responseString, publisherIds);
            return new ConsentInfoUpdateResponse(true, UPDATE_SUCCESS);
        } else {
            return new ConsentInfoUpdateResponse(
                false, urlConnection.getResponseMessage());
        }
    } catch (Exception e) {
        return new ConsentInfoUpdateResponse(false, e.getLocalizedMessage());
    }
}
 

@Override
protected Future<OperationResult<List<CommonCredential>>> processReadCredentials(final DeviceKey key, final Span span) {

    return this.store.getCredentials(key, span.context())
            .<OperationResult<List<CommonCredential>>>map(r -> {

                if (r.isPresent()) {
                    var result = r.get();
                    return OperationResult.ok(
                            HTTP_OK,
                            result.getCredentials(),
                            this.ttl,
                            result.getResourceVersion());
                } else {
                    return empty(HTTP_NOT_FOUND);
                }

            })

            .otherwise(err -> empty(HttpURLConnection.HTTP_INTERNAL_ERROR));

}
 

/**
 * Deletes the container if it exists using the specified request options and operation context.
 * 
 * @param accessCondition
 *            An {@link AccessCondition} object that represents the access conditions for the container.
 * @param options
 *            A {@link BlobRequestOptions} object that specifies any additional options for the request. Specifying
 *            <code>null</code> will use the default request options from the associated service client (
 *            {@link CloudBlobClient}).
 * @param opContext
 *            An {@link OperationContext} object that represents the context for the current operation. This object
 *            is used to track requests to the storage service, and to provide additional runtime information about
 *            the operation.
 * 
 * @return <code>true</code> if the container existed and was deleted; otherwise, <code>false</code>.
 * 
 * @throws StorageException
 *             If a storage service error occurred.
 */
@DoesServiceRequest
public boolean deleteIfExists(AccessCondition accessCondition, BlobRequestOptions options,
        OperationContext opContext) throws StorageException {
    options = BlobRequestOptions.populateAndApplyDefaults(options, BlobType.UNSPECIFIED, this.blobServiceClient);

    boolean exists = this.exists(true /* primaryOnly */, accessCondition, options, opContext);
    if (exists) {
        try {
            this.delete(accessCondition, options, opContext);
            return true;
        }
        catch (StorageException e) {
            if (e.getHttpStatusCode() == HttpURLConnection.HTTP_NOT_FOUND
                    && StorageErrorCodeStrings.CONTAINER_NOT_FOUND.equals(e.getErrorCode())) {
                return false;
            }
            else {
                throw e;
            }
        }
    }
    else {
        return false;
    }
}
 

/**
 * Verifies that an authentication request is sent via the vert.x event bus and the failure reply,
 * containing an invalid error code, is passed on to the handler of the <code>authenticate</code> method
 * as an internal error.
 *
 * @param ctx The vert.x test context.
 */
@Test
public void testAuthenticateFailureInvalidStatusCode(final VertxTestContext ctx) {

    final String failureMessage = "failureMessage";

    authRequestConsumer = vertx.eventBus().consumer(AuthenticationConstants.EVENT_BUS_ADDRESS_AUTHENTICATION_IN, message -> {
        message.fail(200, failureMessage);
    });

    final EventBusAuthenticationService eventBusAuthService = new EventBusAuthenticationService(vertx, authTokenHelperForValidating);
    eventBusAuthService.authenticate(new JsonObject(), ctx.failing(t -> {
        ctx.verify(() -> {
            assertThat(t).isInstanceOf(ServerErrorException.class);
            assertThat(((ServiceInvocationException) t).getErrorCode()).isEqualTo(HttpURLConnection.HTTP_INTERNAL_ERROR);
            assertThat(t.getMessage()).isEqualTo(failureMessage);
        });
        ctx.completeNow();
    }));
}
 

/**
 * Retrieves {@link java.io.InputStream} of image by URI (image is located in the network).
 *
 * @param imageUri Image URI
 * @param extra    Auxiliary object which was passed to {@link com.nostra13.universalimageloader.core.DisplayImageOptions.Builder#extraForDownloader(Object)
 *                 DisplayImageOptions.extraForDownloader(Object)}; can be null
 * @return {@link java.io.InputStream} of image
 * @throws java.io.IOException if some I/O error occurs during network request or if no InputStream could be created for
 *                     URL.
 */
protected InputStream getStreamFromNetwork(String imageUri, Object extra) throws IOException {
	HttpURLConnection conn = createConnection(imageUri, extra);

	int redirectCount = 0;
	while (conn.getResponseCode() / 100 == 3 && redirectCount < MAX_REDIRECT_COUNT) {
		conn = createConnection(conn.getHeaderField("Location"), extra);
		redirectCount++;
	}

	InputStream imageStream;
	try {
		imageStream = conn.getInputStream();
	} catch (IOException e) {
		// Read all data to allow reuse connection (http://bit.ly/1ad35PY)
		IoUtils.readAndCloseStream(conn.getErrorStream());
		throw e;
	}
	return new ContentLengthInputStream(new BufferedInputStream(imageStream, BUFFER_SIZE), conn.getContentLength());
}
 

/**
 * @param fileUrl
 *            文件来源地址
 * @param savePath
 *            文件保存地址
 * @return
 */
public static boolean saveUrlAs(String fileUrl, String savePath) {
	try {
		URL url = new URL(fileUrl);
		HttpURLConnection connection = (HttpURLConnection) url.openConnection();
		DataInputStream in = new DataInputStream(connection.getInputStream());
		DataOutputStream out = new DataOutputStream(new FileOutputStream(savePath));
		byte[] buffer = new byte[4096];
		int count = 0;
		while ((count = in.read(buffer)) > 0) {
			out.write(buffer, 0, count);
		}
		out.close();
		in.close();
		connection.disconnect();
		return true;

	} catch (Exception e) {
		return false;
	}
}
 

protected InputStream getStreamFromNetwork(String s, Object obj)
{
    HttpURLConnection httpurlconnection = createConnection(s, obj);
    for (int i = 0; httpurlconnection.getResponseCode() / 100 == 3 && i < 5; i++)
    {
        httpurlconnection = createConnection(httpurlconnection.getHeaderField("Location"), obj);
    }

    InputStream inputstream;
    try
    {
        inputstream = httpurlconnection.getInputStream();
    }
    catch (IOException ioexception)
    {
        IoUtils.readAndCloseStream(httpurlconnection.getErrorStream());
        throw ioexception;
    }
    return new ContentLengthInputStream(new BufferedInputStream(inputstream, 32768), httpurlconnection.getContentLength());
}
 

@Override
protected ScepHttpResponse httpGet(String url) throws ScepClientException {
  Args.notNull(url, "url");
  try {
    HttpURLConnection httpConn = openHttpConn(new URL(url));
    if (httpConn instanceof HttpsURLConnection) {
      if (sslSocketFactory != null) {
        ((HttpsURLConnection) httpConn).setSSLSocketFactory(sslSocketFactory);
      }
      if (hostnameVerifier != null) {
        ((HttpsURLConnection) httpConn).setHostnameVerifier(hostnameVerifier);
      }
    }

    httpConn.setRequestMethod("GET");
    return parseResponse(httpConn);
  } catch (IOException ex) {
    throw new ScepClientException(ex);
  }
}
 

public static String getNestedURL(String stringUrl) throws MalformedURLException {
	URL url = new URL(stringUrl);
	try {
		HttpURLConnection con = (HttpURLConnection) url.openConnection();
		con.setInstanceFollowRedirects(false);
		con.setRequestProperty("User-Agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/62.0.3202.94 Safari/537.36");
		con.addRequestProperty("Accept-Language", "en-US,en;q=0.8");
		con.addRequestProperty("Referer", "https://www.google.com/");
		con.connect();
		int resultCode = con.getResponseCode();
		if (resultCode == HttpURLConnection.HTTP_SEE_OTHER
				|| resultCode == HttpURLConnection.HTTP_MOVED_PERM
				|| resultCode == HttpURLConnection.HTTP_MOVED_TEMP) {
			String Location = con.getHeaderField("Location");
			if (Location.startsWith("/")) {
				Location = url.getProtocol() + "://" + url.getHost() + Location;
			}
			return getNestedURL(Location);
		}
	} catch (Exception e) {
		System.out.println(e.getMessage());
	}
	return url.toString();
}
 

/**
 * 从网络Url中下载文件
 *
 * @param urlStr 地址
 * @param fileName 文件名称
 * @param savePath 保存路径
 * @throws IOException IO异常
 */
public static void downLoadFromUrl(String urlStr, String fileName, String savePath) throws IOException {
    //文件保存位置
    File saveDir = new File(savePath);
    if (!saveDir.exists()) {
        if (!saveDir.mkdirs()) {
            log.info(savePath + " 文件路径不存在，AGT进行创建失败，请检查是否有权限");
            return;
        }
    }
    File file = new File(saveDir + File.separator + fileName);
    if(file.exists()){
        log.info("配置文件已经存在不进行下载，URL:" + urlStr + ",路径：" + savePath + ",文件名：" + fileName);
        return;
    }

    URL url = new URL(urlStr);
    HttpURLConnection conn = (HttpURLConnection) url.openConnection();
    //设置超时间为30秒
    conn.setConnectTimeout(30 * 1000);
    //防止屏蔽程序抓取而返回403错误
    conn.setRequestProperty("User-Agent", "Mozilla/4.0 (compatible; MSIE 5.0; Windows NT; DigExt)");

    //得到输入流
    InputStream inputStream = conn.getInputStream();
    //获取自己数组
    byte[] getData = readInputStream(inputStream);

    FileOutputStream fos = new FileOutputStream(file);
    fos.write(getData);
    fos.close();
    inputStream.close();
    log.info("下载配置文件成功，URL:" + url + ",路径：" + savePath + ",文件名：" + fileName);
}
 

/**
 * Initializes an {@link HttpEntity} from the given {@link HttpURLConnection}.
 *
 * @param connection
 * @return an HttpEntity populated with data from <code>connection</code>.
 */
private static HttpEntity entityFromConnection(HttpURLConnection connection) {
    BasicHttpEntity entity = new BasicHttpEntity();
    InputStream inputStream;
    try {
        inputStream = connection.getInputStream();
    } catch (IOException ioe) {
        inputStream = connection.getErrorStream();
    }
    entity.setContent(inputStream);
    entity.setContentLength(connection.getContentLength());
    entity.setContentEncoding(connection.getContentEncoding());
    entity.setContentType(connection.getContentType());
    return entity;
}
 

@Test
public void testWebAppProxyInStandAloneMode() throws YarnException,
    InterruptedException, IOException {
  conf.setBoolean(YarnConfiguration.AUTO_FAILOVER_ENABLED, false);
  WebAppProxyServer webAppProxyServer = new WebAppProxyServer();
  try {
    conf.set(YarnConfiguration.PROXY_ADDRESS, "0.0.0.0:9099");
    cluster.init(conf);
    cluster.start();
    getAdminService(0).transitionToActive(req);
    assertFalse("RM never turned active", -1 == cluster.getActiveRMIndex());
    verifyConnections();
    webAppProxyServer.init(conf);

    // Start webAppProxyServer
    Assert.assertEquals(STATE.INITED, webAppProxyServer.getServiceState());
    webAppProxyServer.start();
    Assert.assertEquals(STATE.STARTED, webAppProxyServer.getServiceState());

    // send httpRequest with fakeApplicationId
    // expect to get "Not Found" response and 404 response code
    URL wrongUrl = new URL("http://0.0.0.0:9099/proxy/" + fakeAppId);
    HttpURLConnection proxyConn = (HttpURLConnection) wrongUrl
        .openConnection();

    proxyConn.connect();
    verifyResponse(proxyConn);

    explicitFailover();
    verifyConnections();
    proxyConn.connect();
    verifyResponse(proxyConn);
  } finally {
    webAppProxyServer.stop();
  }
}
 

@NonNull private Response makeRequest(Chain chain, Request request) throws IOException {
    Response r = addHeaderAndProceedWithChain(chain, request);
    if (r.code() == HttpURLConnection.HTTP_FORBIDDEN || r.code() == HttpURLConnection.HTTP_UNAUTHORIZED) {
        mTokenStorage.forceExpireToken();
        // throw an IOException, so that this request will be retried
        throw new IOException("Token problem, throwing to retry");
    }
    return r;
}
 

/**
 * 下载对应url下的apk文件
 */
private void downloadAPK() throws Exception {
    FileOutputStream fos = new FileOutputStream(apkFile);
    String path = updateInfo.apkUrl;
    URL url = new URL(path);
    HttpURLConnection conn = (HttpURLConnection) url.openConnection();
    conn.setConnectTimeout(5000);
    conn.setReadTimeout(5000);
    conn.setRequestMethod("GET");
    conn.connect();

    if (conn.getResponseCode() == 200) {
        InputStream is = conn.getInputStream();
        dialog.setMax(conn.getContentLength());
        byte[] buffer = new byte[1024];
        int len;
        while ((len = is.read(buffer)) != -1) {
            fos.write(buffer, 0, len);
            dialog.incrementProgressBy(len);
            Thread.sleep(2);
        }
        // 暂且使用throws的方式处理异常了
        is.close();
        fos.close();
    } else {
        handler.sendEmptyMessage(WHAT_DOWNLOAD_FAIL);
    }
    // 关闭连接
    conn.disconnect();
}
 

private HttpURLConnection openConnection ( final URL url ) throws IOException
{
    final String host = this.properties.getProperty ( "local.proxy.host" );
    final String port = this.properties.getProperty ( "local.proxy.port" );

    if ( host != null && port != null && !host.isEmpty () && !port.isEmpty () )
    {
        final Proxy proxy = new Proxy ( Type.HTTP, new InetSocketAddress ( host, Integer.parseInt ( port ) ) );
        return (HttpURLConnection)url.openConnection ( proxy );
    }
    else
    {
        return (HttpURLConnection)url.openConnection ();
    }
}
 

@Test
@Category({ DevFabricTests.class, DevStoreTests.class })
public void testCloudBlobContainerInvalidMetadata() throws StorageException{
    // test client-side fails correctly
    testMetadataFailures(this.container, null, "value1", true);
    testMetadataFailures(this.container, "", "value1", true);
    testMetadataFailures(this.container, " ", "value1", true);
    testMetadataFailures(this.container, "\n \t", "value1", true);

    testMetadataFailures(this.container, "key1", null, false);
    testMetadataFailures(this.container, "key1", "", false);
    testMetadataFailures(this.container, "key1", " ", false);
    testMetadataFailures(this.container, "key1", "\n \t", false);

    // test client can get empty metadata
    this.container.create();

    OperationContext opContext = new OperationContext();
    opContext.getSendingRequestEventHandler().addListener(new StorageEvent<SendingRequestEvent>() {
        // insert a metadata element with an empty value
        @Override
        public void eventOccurred(SendingRequestEvent eventArg) {
            HttpURLConnection request = (HttpURLConnection) eventArg.getConnectionObject();
            request.setRequestProperty(Constants.HeaderConstants.PREFIX_FOR_STORAGE_METADATA + "key1", "");
        }
    });
    this.container.uploadMetadata(null, null, opContext);

    this.container.downloadAttributes();
    assertEquals(1, this.container.getMetadata().size());
    assertEquals("", this.container.getMetadata().get("key1"));
}
 

public static boolean isAccessible(URL url, boolean isChipsterServer) throws IOException {
	// check the URL
	HttpURLConnection connection = (HttpURLConnection)url.openConnection();
	if (isChipsterServer) {
		KeyAndTrustManager.configureForChipsterCertificate(connection);
	} else {
		KeyAndTrustManager.configureForCACertificates(connection);
	}
	connection.setConnectTimeout(HTTP_TIMEOUT_MILLISECONDS);
	connection.connect() ; 
	return connection.getResponseCode() == HttpURLConnection.HTTP_OK;
}
 

private HttpStatusCodes checkStatus(HttpURLConnection connection) throws IOException {
	HttpStatusCodes httpStatusCode = HttpStatusCodes.fromStatusCode(connection.getResponseCode());
	if (400 <= httpStatusCode.getStatusCode() && httpStatusCode.getStatusCode() <= 599) {
		throw new RuntimeException("Http Connection failed with status " + httpStatusCode.getStatusCode() + " "
				+ httpStatusCode.toString());
	}
	return httpStatusCode;
}
 

public void upgrade() {
	// get upgrade dialog
	getUpgradeDialog();
	// initialize 
	boolean[] failed = new boolean[] {false};
	int responseCode = 0; // code 200 = "OK"
	// look for upgrade tracker.jar
		final String jarFileName = "tracker-"+Tracker.newerVersion+".jar"; //$NON-NLS-1$ //$NON-NLS-2$
	final String upgradeURL = ResourceLoader.getString("https://physlets.org/tracker/upgradeURL.txt"); //$NON-NLS-1$
	if (upgradeURL!=null && Tracker.trackerHome!=null) {
 		String upgradeFile = upgradeURL.trim()+jarFileName;
 		try {
 	    URL url = new URL(upgradeFile);
 	    HttpURLConnection huc = (HttpURLConnection)url.openConnection();
 	    responseCode = huc.getResponseCode();
   	} catch (Exception ex) {
   	}
	}
	if (responseCode!=200) { 
		// jar file not found
		failed[0] = true;
	}
	else if (OSPRuntime.isWindows()) {
		upgradeWindows(failed);
	}
	else if (OSPRuntime.isMac()) { // OSX
		upgradeOSX(failed);
	}
	else if (OSPRuntime.isLinux()) {
		upgradeLinux(failed);
	}
	if (failed[0]) {
		// close upgrade dialog and display Tracker web site
		closeUpgradeDialog();
 		String websiteurl = "https://"+Tracker.trackerWebsite; //$NON-NLS-1$
 		OSPDesktop.displayURL(websiteurl);
	}

}
 

/**
 * Forwards a message to the AMQP Messaging Network.
 *
 * @param ctx The context in which the MQTT message has been published.
 * @param resource The resource that the message should be forwarded to.
 * @param message The message to send.
 * @return A future indicating the outcome of the operation.
 *         <p>
 *         The future will succeed if the message has been forwarded successfully.
 *         Otherwise the future will fail with a {@link ServiceInvocationException}.
 * @throws NullPointerException if any of context, resource or payload is {@code null}.
 * @throws IllegalArgumentException if the payload is empty.
 */
public final Future<Void> uploadMessage(
        final MqttContext ctx,
        final ResourceIdentifier resource,
        final MqttPublishMessage message) {

    Objects.requireNonNull(ctx);
    Objects.requireNonNull(resource);
    Objects.requireNonNull(message);

    switch (MetricsTags.EndpointType.fromString(resource.getEndpoint())) {
    case TELEMETRY:
        return uploadTelemetryMessage(
                ctx,
                resource.getTenantId(),
                resource.getResourceId(),
                message.payload());
    case EVENT:
        return uploadEventMessage(
                ctx,
                resource.getTenantId(),
                resource.getResourceId(),
                message.payload());
    case COMMAND:
        return uploadCommandResponseMessage(ctx, resource);
    default:
        return Future
                .failedFuture(new ClientErrorException(HttpURLConnection.HTTP_BAD_REQUEST, "unsupported endpoint"));
    }

}
 

/**
 * Execute the given request through a standard J2SE HttpURLConnection.
 * <p>This method implements the basic processing workflow:
 * The actual work happens in this class's template methods.
 * @see #openConnection
 * @see #prepareConnection
 * @see #writeRequestBody
 * @see #validateResponse
 * @see #readResponseBody
 */
@Override
protected RemoteInvocationResult doExecuteRequest(
		HttpInvokerClientConfiguration config, ByteArrayOutputStream baos)
		throws IOException, ClassNotFoundException {

	HttpURLConnection con = openConnection(config);
	prepareConnection(con, baos.size());
	writeRequestBody(config, con, baos);
	validateResponse(config, con);
	InputStream responseBody = readResponseBody(config, con);

	return readRemoteInvocationResult(responseBody, config.getCodebaseUrl());
}
 

static private void setHeaders(HttpURLConnection conn, List<String> headers, String encoding) {
    if (null != headers) {
        for (Iterator<String> iter = headers.iterator(); iter.hasNext(); ) {
            conn.addRequestProperty(iter.next(), iter.next());
        }
    }
    conn.addRequestProperty("Client-Version", MQVersion.getVersionDesc(MQVersion.CURRENT_VERSION));
    conn.addRequestProperty("Content-Type", "application/x-www-form-urlencoded;charset=" + encoding);

    String ts = String.valueOf(System.currentTimeMillis());
    conn.addRequestProperty("Metaq-Client-RequestTS", ts);
}
 

HttpURLConnection open(URL url, Proxy proxy) {
  String protocol = url.getProtocol();
  OkHttpClient copy = client.clone();

  if (protocol.equals("http")) return new HttpURLConnectionImpl(url, copy);
  if (protocol.equals("https")) return new HttpsURLConnectionImpl(url, copy);
  throw new IllegalArgumentException("Unexpected protocol: " + protocol);
}