/*
 * Copyright 2002-2016 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.springframework.http.client.reactive;

import java.net.URI;
import java.time.Duration;
import java.util.Collection;
import java.util.Optional;

import io.netty.handler.codec.http.cookie.Cookie;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.io.buffer.Buffer;
import reactor.io.netty.http.HttpClient;

import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferAllocator;
import org.springframework.http.HttpCookie;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.ResponseCookie;

/**
 * {@link ClientHttpRequest} implementation for the Reactor Net HTTP client
 *
 * @author Brian Clozel
 * @see HttpClient
 */
public class ReactorClientHttpRequest extends AbstractClientHttpRequest {

	private final DataBufferAllocator allocator;

	private final HttpMethod httpMethod;

	private final URI uri;

	private final HttpClient httpClient;

	private Flux<Buffer> body;


	public ReactorClientHttpRequest(HttpMethod httpMethod, URI uri, HttpClient httpClient, HttpHeaders headers,
			DataBufferAllocator allocator) {
		super(headers);
		this.allocator = allocator;
		this.httpMethod = httpMethod;
		this.uri = uri;
		this.httpClient = httpClient;
	}

	@Override
	public DataBufferAllocator allocator() {
		return this.allocator;
	}

	@Override
	public HttpMethod getMethod() {
		return this.httpMethod;
	}

	@Override
	public URI getURI() {
		return this.uri;
	}

	/**
	 * Set the body of the message to the given {@link Publisher}.
	 *
	 * <p>Since the HTTP channel is not yet created when this method
	 * is called, the {@code Mono<Void>} return value completes immediately.
	 * For an event that signals that we're done writing the request, check the
	 * {@link #execute()} method.
	 *
	 * @return a publisher that completes immediately.
	 * @see #execute()
	 */
	@Override
	public Mono<Void> setBody(Publisher<DataBuffer> body) {

		this.body = Flux.from(body).map(b -> new Buffer(b.asByteBuffer()));
		return Mono.empty();
	}

	@Override
	public Mono<ClientHttpResponse> execute() {

		return this.httpClient.request(new io.netty.handler.codec.http.HttpMethod(httpMethod.toString()), uri.toString(),
				channel -> {
					// see https://github.com/reactor/reactor-io/pull/8
					if (body == null) {
						channel.removeTransferEncodingChunked();
					}
					return applyBeforeCommit()
							.after(() -> {
								getHeaders().entrySet().stream().forEach(e ->
										channel.headers().set(e.getKey(), e.getValue()));
								getCookies().values().stream().flatMap(Collection::stream).forEach(cookie ->
										channel.addCookie(cookie.getName(), new NettyCookie(cookie)));
								return Mono.empty();
							})
							.after(() -> {
								if (body != null) {
									return channel.send(body);
								}
								else {
									return channel.writeHeaders();
								}
							});
				})
				.map(httpChannel -> new ReactorClientHttpResponse(httpChannel, allocator));
	}

	private final static class NettyCookie implements Cookie {

		private final HttpCookie httpCookie;


		public NettyCookie(HttpCookie httpCookie) {
			this.httpCookie = httpCookie;
		}

		@Override
		public String name() {
			return this.httpCookie.getName();
		}

		@Override
		public String value() {
			return this.httpCookie.getValue();
		}

		@Override
		public boolean isHttpOnly() {
			return true;
		}

		@Override
		public long maxAge() {
			return -1;
		}

		@Override
		public String domain() {
			return null;
		}

		@Override
		public String path() {
			return null;
		}

		@Override
		public void setValue(String value) {
			throw new UnsupportedOperationException("Read-Only Cookie");
		}

		@Override
		public boolean wrap() {
			return false;
		}

		@Override
		public void setWrap(boolean wrap) {
			throw new UnsupportedOperationException("Read-Only Cookie");
		}

		@Override
		public void setDomain(String domain) {
			throw new UnsupportedOperationException("Read-Only Cookie");
		}

		@Override
		public void setPath(String path) {
			throw new UnsupportedOperationException("Read-Only Cookie");
		}

		@Override
		public void setMaxAge(long maxAge) {
			throw new UnsupportedOperationException("Read-Only Cookie");
		}

		@Override
		public void setSecure(boolean secure) {
			throw new UnsupportedOperationException("Read-Only Cookie");
		}

		@Override
		public void setHttpOnly(boolean httpOnly) {
			throw new UnsupportedOperationException("Read-Only Cookie");
		}

		@Override
		public int compareTo(Cookie o) {
			return httpCookie.getName().compareTo(o.name());
		}

		@Override
		public boolean isSecure() {
			return false;
		}
	}

}


@@ -3580,13 +3580,12 @@
 nel.
-write
+send
 Head

fix change issue on HttpChannel#writeHeaders to sendHeaders