/*
 * The contents of this file are subject to the terms of the Common Development and
 * Distribution License (the License). You may not use this file except in compliance with the
 * License.
 *
 * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the
 * specific language governing permission and limitations under the License.
 *
 * When distributing Covered Software, include this CDDL Header Notice in each file and include
 * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL
 * Header, with the fields enclosed by brackets [] replaced by your own identifying
 * information: "Portions Copyright [year] [name of copyright owner]".
 *
 * Copyright 2016 ForgeRock AS.
 */
package org.forgerock.opendj.reactive;

import static com.forgerock.reactive.RxJavaStreams.*;
import static org.forgerock.util.Utils.closeSilently;

import java.util.concurrent.atomic.AtomicReference;

import org.forgerock.opendj.ldap.Connection;
import org.forgerock.opendj.ldap.IntermediateResponseHandler;
import org.forgerock.opendj.ldap.LDAPConnectionFactory;
import org.forgerock.opendj.ldap.LdapException;
import org.forgerock.opendj.ldap.SearchResultHandler;
import org.forgerock.opendj.ldap.requests.CompareRequest;
import org.forgerock.opendj.ldap.requests.ExtendedRequest;
import org.forgerock.opendj.ldap.requests.Request;
import org.forgerock.opendj.ldap.requests.SearchRequest;
import org.forgerock.opendj.ldap.responses.IntermediateResponse;
import org.forgerock.opendj.ldap.responses.Response;
import org.forgerock.opendj.ldap.responses.Result;
import org.forgerock.opendj.ldap.responses.SearchResultEntry;
import org.forgerock.opendj.ldap.responses.SearchResultReference;
import org.forgerock.opendj.ldif.ChangeRecord;
import org.forgerock.util.promise.ExceptionHandler;
import org.forgerock.util.promise.ResultHandler;
import org.forgerock.util.promise.RuntimeExceptionHandler;

import com.forgerock.reactive.ReactiveHandler;
import com.forgerock.reactive.Single;
import com.forgerock.reactive.Stream;

import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableEmitter.BackpressureMode;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.functions.Action;

/** Forward {@link Request} to another server reached through the {@link LDAPConnectionFactory}. */
final class ProxyToHandler implements ReactiveHandler<LDAPClientConnection2, Request, Stream<Response>> {

    private final LDAPConnectionFactory connectionFactory;

    ProxyToHandler(final LDAPConnectionFactory connectionFactory) {
        this.connectionFactory = connectionFactory;
    }

    @Override
    public Single<Stream<Response>> handle(final LDAPClientConnection2 context, final Request request) {
        return singleFromPublisher(Flowable.create(new FlowableOnSubscribe<Stream<Response>>() {
            @Override
            public void subscribe(final FlowableEmitter<Stream<Response>> emitter) throws Exception {
                final AtomicReference<Connection> connectionHolder = new AtomicReference<Connection>();

                connectionFactory.getConnectionAsync().thenOnResult(new ResultHandler<Connection>() {
                    @Override
                    public void handleResult(final Connection connection) {
                        connectionHolder.set(connection);
                        emitter.onNext(executeRequest(connection, request));
                        emitter.onComplete();
                    }
                }).thenOnException(new ExceptionHandler<LdapException>() {
                    @Override
                    public void handleException(LdapException exception) {
                        emitter.onError(exception);
                    }
                }).thenOnRuntimeException(new RuntimeExceptionHandler() {
                    @Override
                    public void handleRuntimeException(RuntimeException exception) {
                        emitter.onError(exception);
                    }
                });
            }
        }, BackpressureMode.ERROR));
    }

    private Stream<Response> executeRequest(final Connection connection, final Request request) {
        return streamFromPublisher(Flowable.create(new FlowableOnSubscribe<Response>() {
            @Override
            public void subscribe(FlowableEmitter<Response> emitter) throws Exception {
                // add/delete/modifyDN/modifyReq
                final PublisherAdaptor adapter = new PublisherAdaptor(emitter);
                if (request instanceof ChangeRecord) {
                    connection.applyChangeAsync((ChangeRecord) request, adapter).thenOnResult(adapter)
                            .thenOnException(adapter).thenOnRuntimeException(adapter);
                } else if (request instanceof CompareRequest) {
                    connection.compareAsync((CompareRequest) request, adapter).thenOnResult(adapter)
                            .thenOnException(adapter).thenOnRuntimeException(adapter);
                } else if (request instanceof ExtendedRequest) {
                    connection.compareAsync((CompareRequest) request, adapter).thenOnResult(adapter)
                            .thenOnException(adapter).thenOnRuntimeException(adapter);
                } else if (request instanceof SearchRequest) {
                    connection.searchAsync((SearchRequest) request, adapter).thenOnResult(adapter)
                            .thenOnException(adapter).thenOnRuntimeException(adapter);
                } else {
                    emitter.onError(new IllegalArgumentException("Unsupported request type"));
                }
            }
        }, BackpressureMode.ERROR).doAfterTerminate(new Action() {
            @Override
            public void run() throws Exception {
                closeSilently(connection);
            }
        }));
    }

    /** Adaptor forwarding events from the SDK handler to the emitter. */
    private final class PublisherAdaptor implements IntermediateResponseHandler, SearchResultHandler,
            ResultHandler<Result>, ExceptionHandler<Exception>, RuntimeExceptionHandler {

        private final FlowableEmitter<Response> emitter;

        PublisherAdaptor(final FlowableEmitter<Response> emitter) {
            this.emitter = emitter;
        }

        private boolean handle(final Response response) {
            emitter.onNext(response);
            return true;
        }

        @Override
        public boolean handleEntry(SearchResultEntry entry) {
            return handle(entry);
        }

        @Override
        public boolean handleReference(SearchResultReference reference) {
            return handle(reference);
        }

        @Override
        public boolean handleIntermediateResponse(IntermediateResponse response) {
            return handle(response);
        }

        @Override
        public void handleResult(Result result) {
            if (result != null) {
                handle(result);
            }
            emitter.onComplete();
        }

        @Override
        public void handleRuntimeException(RuntimeException exception) {
            emitter.onError(exception);
        }

        @Override
        public void handleException(Exception exception) {
            emitter.onError(exception);
        }
    }
}