package com.microsoft.rest.v2.http;

import io.netty.buffer.ByteBuf;
import io.netty.handler.codec.http.HttpMethod;
import io.reactivex.netty.protocol.http.client.HttpClientRequest;
import io.reactivex.netty.protocol.http.client.HttpClientResponse;
import rx.Observable;
import rx.Single;
import rx.functions.Func1;

import java.net.URI;
import java.net.URISyntaxException;

public class RxNettyClient extends HttpClient {
    @Override
    public Single<? extends HttpResponse> sendRequestAsync(HttpRequest request) {
        URI uri;
        try {
            uri = new URI(request.getURL());
        } catch (URISyntaxException e) {
            return Single.error(e);
        }

        HttpClientRequest<ByteBuf, ByteBuf> rxnReq = io.reactivex.netty.protocol.http.client.HttpClient
                .newClient(uri.getHost(), 80)
                .createRequest(HttpMethod.valueOf(request.getHttpMethod()), uri.toASCIIString());

        for (HttpHeader header : request.getHeaders()) {
            rxnReq = rxnReq.addHeader(header.getName(), header.getValue());
        }

        String mimeType = request.getMIMEType();
        if (mimeType != null) {
            rxnReq = rxnReq.addHeader("Content-Type", mimeType);
        }

        String body = request.getBody();
        Observable<HttpClientResponse<ByteBuf>> obsResponse = rxnReq;
        if (body != null) {
            if (rxnReq.getHeader("Content-Length") == null) {
                rxnReq = rxnReq.addHeader("Content-Length", body.length());
            }

            obsResponse = rxnReq.writeStringContent(Observable.just(body));
        }

        return obsResponse
                .map(new Func1<HttpClientResponse<ByteBuf>, HttpResponse>() {
                    @Override
                    public HttpResponse call(HttpClientResponse<ByteBuf> rxnRes) {
                        return new RxNettyResponse(rxnRes);
                    }
                })
                .toSingle();
    }
}

@@ -366,16 +366,116 @@
 ception;
+%0Aimport java.util.Collections;%0Aimport java.util.HashMap;%0Aimport java.util.Map;%0Aimport java.util.Set;
 %0A%0Apublic
@@ -788,249 +788,63 @@
     
-HttpClientRequest%3CByteBuf, ByteBuf%3E rxnReq = io.reactivex.netty.protocol.http.client.HttpClient%0A                .newClient(uri.getHost(), 80)%0A                .createRequest(HttpMethod.valueOf(request.getHttpMethod()), uri.toASCIIString()
+Map%3CString, Set%3CObject%3E%3E rxnHeaders = new HashMap%3C%3E(
 );%0A
-%0A
     
@@ -911,38 +911,27 @@
      rxn
-Req = rxnReq.add
 Header
+s.put
 (header.
@@ -941,16 +941,46 @@
 Name(), 
+Collections.%3CObject%3Esingleton(
 header.g
@@ -989,16 +989,17 @@
 Value())
+)
 ;%0A      
@@ -1099,38 +1099,27 @@
      rxn
-Req = rxnReq.add
 Header
+s.put
 (%22Conten
@@ -1127,16 +1127,46 @@
 -Type%22, 
+Collections.%3CObject%3Esingleton(
 mimeType
@@ -1166,16 +1166,17 @@
 imeType)
+)
 ;%0A      
@@ -1233,19 +1233,155 @@
     
-Observable%3C
+if (body != null) %7B%0A            rxnHeaders.put(%22Content-Length%22, Collections.%3CObject%3Esingleton(String.valueOf(body.length())));%0A        %7D%0A%0A        
 Http
@@ -1388,30 +1388,29 @@
 ClientRe
-sponse
+quest
 %3CByteBuf
 %3E%3E obsRe
@@ -1405,213 +1405,361 @@
 eBuf
-%3E%3E obsResponse = rxnReq;%0A        if (body != null) %7B%0A            if (rxnReq.getHeader(%22Content-Length%22) == null) %7B%0A                rxnReq = rxnReq.addHeader(%22Content-Length%22, body.length());%0A            %7D%0A
+, ByteBuf%3E rxnReq = io.reactivex.netty.protocol.http.client.HttpClient%0A                .newClient(uri.getHost(), 80)%0A                .createRequest(HttpMethod.valueOf(request.getHttpMethod()), uri.toASCIIString())%0A                .addHeaders(rxnHeaders);%0A%0A        Observable%3CHttpClientResponse%3CByteBuf%3E%3E obsResponse = rxnReq;%0A        if (body != null) %7B
 %0A   

Use addHeaders to reduce RxNetty request copying