/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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.apache.camel.quarkus.component.bean;

import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicInteger;

import io.quarkus.runtime.annotations.RegisterForReflection;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;
import jakarta.inject.Named;
import org.apache.camel.AsyncCallback;
import org.apache.camel.AsyncProcessor;
import org.apache.camel.BindToRegistry;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.bean.BeanProcessor;
import org.apache.camel.support.DefaultExchange;

/**
 * A {@link RouteBuilder} instantiated by Camel (not by Arc).
 */
@ApplicationScoped
public class BeanRoutes extends RouteBuilder {

    static final AtomicInteger CONFIGURE_COUNTER = new AtomicInteger(0);

    @Inject
    @Named("collected-names")
    Map<String, List<String>> collectedNames;

    @Override
    public void addRoutesToCamelContext(CamelContext context) throws Exception {
        CONFIGURE_COUNTER.incrementAndGet();
        super.addRoutesToCamelContext(context);
    }

    @Override
    public void configure() {
        from("direct:process-order")
                .setHeader(MyOrderService.class.getName(), MyOrderService::new)
                .split(body().tokenize("@"), BeanRoutes.this::aggregate)
                // each splitted message is then send to this bean where we can process it
                .process(stateless(MyOrderService.class.getName(), "handleOrder"))
                // this is important to end the splitter route as we do not want to do more routing
                // on each splitted message
                .end()
                // after we have splitted and handled each message we want to send a single combined
                // response back to the original caller, so we let this bean build it for us
                // this bean will receive the result of the aggregate strategy: MyOrderStrategy
                .process(stateless(MyOrderService.class.getName(), "buildCombinedResponse"));

        from("direct:named")
                .to("bean:namedBean?method=hello");

        from("direct:identified")
                .to("bean:identifiedBean?method=hello");

        from("direct:method")
                .bean(MyBean.class, "sayHello");

        from("direct:handler")
                .to("bean:withHandler");

        from("direct:handlerOnProxy")
                .to("bean:withHandlerOnProxy");

        from("direct:with-language-param-bindings")
                .routeId("wlpb-route-31wp")
                .to("bean:withLanguageParamBindingsBean");

        from("direct:methodWithExchangeArg")
                .to("bean:namedBean");

        from("direct:completionStageBean")
                .to("bean:completionStageBean?method=hello");

        from("direct:multiArgMethod")
                .to("bean:parametersBean?method=multiArgMethod");

        from("direct:parameterBindingAnnotations")
                .to("bean:parametersBean?method=parameterBindingAnnotations");

        from("direct:parameterLiterals")
                .to("bean:parametersBean?method=parameterLiterals(*, true)");

        from("direct:parameterTypes")
                .to("bean:parametersBean?method=parameterTypes(String.class)");

        from("direct:produceInterface")
                .process(e -> e.getMessage().setBody("produceInterface " + e.getMessage().getBody(String.class)));

        from("direct:propertyInject")
                .toD("bean:${header.beanName}?method=${header.beanMethod}");
    }

    @SuppressWarnings("unchecked")
    public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
        // put order together in old exchange by adding the order from new exchange
        List<String> orders;
        if (oldExchange != null) {
            orders = oldExchange.getIn().getBody(List.class);
        } else {
            orders = new ArrayList<>();
            oldExchange = new DefaultExchange(newExchange.getContext());
            oldExchange.getIn().copyFromWithNewBody(newExchange.getIn(), orders);
        }
        String newLine = newExchange.getIn().getBody(String.class);

        log.debug("Aggregate old orders: " + orders);
        log.debug("Aggregate new order: " + newLine);

        // add orders to the list
        orders.add(newLine);

        // return old as this is the one that has all the orders gathered until now
        return oldExchange;
    }

    private Processor stateless(String header, String method) {
        return new AsyncProcessor() {
            @Override
            public boolean process(Exchange exchange, AsyncCallback callback) {
                return getBeanProcess(exchange).process(exchange, callback);
            }

            @Override
            public CompletableFuture<Exchange> processAsync(Exchange exchange) {
                return getBeanProcess(exchange).processAsync(exchange);
            }

            @Override
            public void process(Exchange exchange) throws Exception {
                getBeanProcess(exchange).process(exchange);
            }

            protected BeanProcessor getBeanProcess(Exchange exchange) {
                BeanProcessor bp = new BeanProcessor(
                        exchange.getIn().getHeader(header),
                        exchange.getContext());
                bp.setMethod(method);
                return bp;
            }
        };
    }

    @RegisterForReflection
    public class MyOrderService {

        private int counter;

        /**
         * We just handle the order by returning a id line for the order
         */
        public String handleOrder(String line) {
            log.debug("HandleOrder: " + line);
            return "(id=" + ++counter + ",item=" + line + ")";
        }

        /**
         * We use the same bean for building the combined response to send
         * back to the original caller
         */
        public Map<String, Object> buildCombinedResponse(List<String> lines) {
            log.debug("BuildCombinedResponse: " + lines);
            Map<String, Object> result = new LinkedHashMap<>();
            result.put("success", true);
            result.put("lines", lines);
            return result;
        }
    }

    @RegisterForReflection
    public static class MyBean {
        /**
         * Just return an hello message.
         */
        public static String sayHello(String body) {
            return "Hello " + body + " from the MyBean";
        }
    }

    @BindToRegistry
    public static WithHandlerBean withHandler() {
        return new WithHandlerBean();
    }

}
