class NettyHttpServiceClient implements ServiceClient {

    private void sendHttpRequest(Operation op) {
        final Object originalBody = op.getBodyRaw();
        try {
            byte[] body = Utils.encodeBody(op, true);
            if (op.getContentLength() > getRequestPayloadSizeLimit()) {
                String error = String.format("Content length %d, limit is %d",
                        op.getContentLength(), getRequestPayloadSizeLimit());
                Exception e = new IllegalArgumentException(error);
                op.setBody(ServiceErrorResponse.create(e, Operation.STATUS_CODE_BAD_REQUEST));
                fail(e, op, originalBody);
                return;
            }

            String pathAndQuery;
            String path = op.getUri().getPath();
            String query = op.getUri().getRawQuery();
            String userInfo = op.getUri().getRawUserInfo();
            path = path == null || path.isEmpty() ? "/" : path;
            if (query != null) {
                pathAndQuery = path + "?" + query;
            } else {
                pathAndQuery = path;
            }

            boolean useHttp2 = op.isConnectionSharing();
            if (this.httpProxy != null || useHttp2 || userInfo != null) {
                pathAndQuery = op.getUri().toString();
            }

            NettyFullHttpRequest request = null;
            HttpMethod method = toHttpMethod(op.getAction());
            if (body == null || body.length == 0) {
                request = new NettyFullHttpRequest(HttpVersion.HTTP_1_1, method, pathAndQuery,
                        Unpooled.EMPTY_BUFFER, false);
            } else {
                ByteBuf content = Unpooled.wrappedBuffer(body, 0, (int) op.getContentLength());
                request = new NettyFullHttpRequest(HttpVersion.HTTP_1_1, method, pathAndQuery,
                        content, false);
            }

            HttpHeaders httpHeaders = request.headers();

            /*
             * NOTE: Pay close attention to calls that access the operation request headers, since
             * they will cause a memory allocation. We avoid the allocation by first checking if
             * the operation has any custom headers to begin with, then we check for the specific
             * header
             */
            boolean hasRequestHeaders = op.hasRequestHeaders();

            if (useHttp2) {
                // when operation is cloned, it may contain original streamId header. remove it.
                if (hasRequestHeaders) {
                    op.getAndRemoveRequestHeaderAsIs(Operation.STREAM_ID_HEADER);
                }
                // We set the operation so that once a streamId is assigned, we can record
                // the correspondence between the streamId and operation: this will let us
                // handle responses properly later.
                request.setOperation(op);
            }

            String pragmaValue = op.getAndRemoveRequestHeaderAsIs(Operation.PRAGMA_HEADER);
            String acceptValue = op.getAndRemoveRequestHeaderAsIs(Operation.ACCEPT_HEADER);
            String authTokenValue = op
                    .getAndRemoveRequestHeaderAsIs(Operation.REQUEST_AUTH_TOKEN_HEADER);

            if (op.isFromReplication() && pragmaValue == null) {
                httpHeaders.add(HttpHeaderNames.PRAGMA, PRAGMA_DIRECTIVE_REPLICATED_ASCII);
            } else if (pragmaValue != null) {
                httpHeaders.add(HttpHeaderNames.PRAGMA, pragmaValue);
            }

            if (op.getTransactionId() != null) {
                httpHeaders.add(TRANSACTION_ID_HEADER_ASCII, op.getTransactionId());
            }

            if (op.getContextId() != null) {
                httpHeaders.add(CONTEXT_ID_HEADER_ASCII, op.getContextId());
            }

            hasRequestHeaders = op.hasRequestHeaders();

            String connectionHeaderValue = op.getRequestHeaderAsIs(Operation.CONNECTION_HEADER);

            boolean isXenonToXenon = op.isFromReplication() || op.isForwarded();
            if (hasRequestHeaders) {
                // remove all headers that we will set explicitly
                op.getAndRemoveRequestHeaderAsIs(Operation.CONTENT_LENGTH_HEADER);
                op.getAndRemoveRequestHeaderAsIs(Operation.CONTENT_TYPE_HEADER);
                op.getAndRemoveRequestHeaderAsIs(Operation.CONNECTION_HEADER);
                op.getAndRemoveRequestHeaderAsIs(Operation.REFERER_HEADER);
                op.getAndRemoveRequestHeaderAsIs(Operation.HOST_HEADER);
                op.getAndRemoveRequestHeaderAsIs(Operation.USER_AGENT_HEADER);
                hasRequestHeaders = op.hasRequestHeaders();
            }

            if (hasRequestHeaders) {
                for (Entry<String, String> nameValue : op.getRequestHeaders().entrySet()) {
                    httpHeaders.add(nameValue.getKey(), nameValue.getValue());
                }
            }

            if (authTokenValue != null) {
                httpHeaders.add(AUTH_TOKEN_HEADER_ASCII, authTokenValue);
            } else {
                AuthorizationContext ctx = op.getAuthorizationContext();
                if (ctx != null && ctx.getToken() != null) {
                    httpHeaders.add(AUTH_TOKEN_HEADER_ASCII, ctx.getToken());
                }
            }

            setAsciiContentType(op, httpHeaders);

            httpHeaders.add(HttpHeaderNames.CONTENT_LENGTH,
                    Long.toString(op.getContentLength()));

            if (op.isKeepAlive()) {
                httpHeaders.add(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
            } else if (connectionHeaderValue != null) {
                // If connection header is specified and keep-alive is not enabled, put back original value
                httpHeaders.add(HttpHeaderNames.CONNECTION, connectionHeaderValue);
            }

            if (op.hasReferer() && !op.isFromReplication()) {
                httpHeaders.add(HttpHeaderNames.REFERER, op.getRefererAsString());
            }

            if (!isXenonToXenon) {
                if (op.getCookies() != null) {
                    String header = CookieJar.encodeCookies(op.getCookies());
                    httpHeaders.set(HttpHeaderNames.COOKIE, header);
                }

                request.headers().set(HttpHeaderNames.USER_AGENT, this.userAgentAscii);
                if (acceptValue == null) {
                    httpHeaders.add(HttpHeaderNames.ACCEPT, DEFAULT_MEDIA_TYPE_ASCII);
                } else {
                    httpHeaders.add(HttpHeaderNames.ACCEPT, acceptValue);
                }

                httpHeaders.add(HttpHeaderNames.HOST, op.getUri().getHost());
            } else {
                if (acceptValue != null) {
                    httpHeaders.add(HttpHeaderNames.ACCEPT, acceptValue);
                }
            }

            if (LOGGER.isLoggable(Level.FINEST)) {
                logRequestFraming(op, request);
            }

            boolean doCookieJarUpdate = !isXenonToXenon;
            op.nestCompletion((o, e) -> {
                if (e != null) {
                    fail(e, op, originalBody);
                    return;
                }

                stopTracking(op);

                if (doCookieJarUpdate) {
                    updateCookieJarFromResponseHeaders(o);
                }

                // After request is sent control is transferred to the
                // NettyHttpServerResponseHandler. The response handler will nest completions
                // and call complete() when response is received, which will invoke this completion
                op.complete();
            });

            op.toggleOption(OperationOption.SOCKET_ACTIVE, true);
            op.getSocketContext().writeHttpRequest(request);
        } catch (Exception e) {
            op.setBody(ServiceErrorResponse.create(e, Operation.STATUS_CODE_BAD_REQUEST,
                    EnumSet.of(ErrorDetail.SHOULD_RETRY)));
            fail(e, op, originalBody);
        }
    }

}