class HttpClient extends AbstractSiteToSiteClient implements PeerStatusProvider {

    @Override
    public Transaction createTransaction(final TransferDirection direction) throws HandshakeException, PortNotRunningException, ProtocolException, UnknownPortException, IOException {
        final int timeoutMillis = (int) config.getTimeout(TimeUnit.MILLISECONDS);

        PeerStatus peerStatus;
        while ((peerStatus = peerSelector.getNextPeerStatus(direction)) != null) {
            logger.debug("peerStatus={}", peerStatus);

            final CommunicationsSession commSession = new HttpCommunicationsSession();
            final String nodeApiUrl = resolveNodeApiUrl(peerStatus.getPeerDescription());
            final StringBuilder clusterUrls = new StringBuilder();
            config.getUrls().forEach(url -> {
                if (clusterUrls.length() > 0) {
                    clusterUrls.append(",");
                    clusterUrls.append(url);
                }
            });
            final Peer peer = new Peer(peerStatus.getPeerDescription(), commSession, nodeApiUrl, clusterUrls.toString());

            final int penaltyMillis = (int) config.getPenalizationPeriod(TimeUnit.MILLISECONDS);
            String portId = config.getPortIdentifier();
            if (StringUtils.isEmpty(portId)) {
                portId = siteInfoProvider.getPortIdentifier(config.getPortName(), direction);
                if (StringUtils.isEmpty(portId)) {
                    peer.close();
                    throw new IOException("Failed to determine the identifier of port " + config.getPortName());
                }
            }

            final SiteToSiteRestApiClient apiClient = new SiteToSiteRestApiClient(config.getSslContext(), config.getHttpProxy(), config.getEventReporter());

            apiClient.setBaseUrl(peer.getUrl());
            apiClient.setConnectTimeoutMillis(timeoutMillis);
            apiClient.setReadTimeoutMillis(timeoutMillis);

            apiClient.setCompress(config.isUseCompression());
            apiClient.setRequestExpirationMillis(config.getIdleConnectionExpiration(TimeUnit.MILLISECONDS));
            apiClient.setBatchCount(config.getPreferredBatchCount());
            apiClient.setBatchSize(config.getPreferredBatchSize());
            apiClient.setBatchDurationMillis(config.getPreferredBatchDuration(TimeUnit.MILLISECONDS));

            final String transactionUrl;
            try {
                transactionUrl = apiClient.initiateTransaction(direction, portId);
                commSession.setUserDn(apiClient.getTrustedPeerDn());
            } catch (final Exception e) {
                apiClient.close();
                logger.warn("Penalizing a peer {} due to {}", peer, e.toString());
                peerSelector.penalize(peer, penaltyMillis);

                // Following exceptions will be thrown even if we tried other peers, so throw it.
                if (e instanceof UnknownPortException
                        || e instanceof PortNotRunningException
                        || e instanceof HandshakeException) {
                    throw e;
                }

                logger.debug("Continue trying other peers...");
                continue;
            }

            // We found a valid peer to communicate with.
            final Integer transactionProtocolVersion = apiClient.getTransactionProtocolVersion();
            final HttpClientTransaction transaction = new HttpClientTransaction(transactionProtocolVersion, peer, direction,
                config.isUseCompression(), portId, penaltyMillis, config.getEventReporter()) {

                @Override
                protected void close() throws IOException {
                    try {
                        super.close();
                    } finally {
                        activeTransactions.remove(this);
                    }
                }
            };

            try {
                transaction.initialize(apiClient, transactionUrl);
            } catch (final Exception e) {
                transaction.error();
                throw e;
            }

            activeTransactions.add(transaction);
            return transaction;
        }

        logger.info("Couldn't find a valid peer to communicate with.");
        return null;
    }

    public  HttpClient(final SiteToSiteClientConfig config);

}