/*
 * Yet Another UserAgent Analyzer
 * Copyright (C) 2013-2023 Niels Basjes
 *
 * 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
 *
 * https://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 nl.basjes.parse.useragent.flink;

import nl.basjes.parse.useragent.annotate.YauaaField;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.LocalEnvironment;
import org.apache.flink.api.java.io.LocalCollectionOutputFormat;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.junit.jupiter.api.Test;

import java.util.ArrayList;
import java.util.List;

import static org.hamcrest.CoreMatchers.hasItems;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;

class TestUserAgentAnalysisMapperInline {
    @Test
    void testInlineDefinitionDataSet() throws Exception {
        ExecutionEnvironment environment = LocalEnvironment.getExecutionEnvironment();

        DataSet<TestRecord> resultDataSet = environment
            .fromElements(
                "Mozilla/5.0 (X11; Linux x86_64) " +
                    "AppleWebKit/537.36 (KHTML, like Gecko) " +
                    "Chrome/48.0.2564.82 Safari/537.36",

                "Mozilla/5.0 (Linux; Android 7.0; Nexus 6 Build/NBD90Z) " +
                    "AppleWebKit/537.36 (KHTML, like Gecko) " +
                    "Chrome/53.0.2785.124 Mobile Safari/537.36"
            )

            .map((MapFunction<String, TestRecord>) TestRecord::new)

            .map(new UserAgentAnalysisMapper<TestRecord>(15000) { // Setting the cacheSize
                @Override
                public String getUserAgentString(TestRecord record) {
                    return record.getUserAgent();
                }

                @SuppressWarnings("unused") // Called via the annotation
                @YauaaField("DeviceClass")
                public void setDeviceClass(TestRecord record, String value) {
                    record.deviceClass = value;
                }

                @SuppressWarnings("unused") // Called via the annotation
                @YauaaField("AgentNameVersion")
                public void setAgentNameVersion(TestRecord record, String value) {
                    record.agentNameVersion = value;
                }
            });

        List<TestRecord> result = new ArrayList<>(5);
        resultDataSet
            .output(new LocalCollectionOutputFormat<>(result));

        environment.execute();

        assertEquals(2, result.size());

        assertThat(result, hasItems(
            new TestRecord(
                "Mozilla/5.0 (X11; Linux x86_64) " +
                    "AppleWebKit/537.36 (KHTML, like Gecko) " +
                    "Chrome/48.0.2564.82 Safari/537.36",
                "Desktop",
                "Chrome 48.0.2564.82",
                null),

            new TestRecord(
                "Mozilla/5.0 (Linux; Android 7.0; Nexus 6 Build/NBD90Z) " +
                    "AppleWebKit/537.36 (KHTML, like Gecko) " +
                    "Chrome/53.0.2785.124 Mobile Safari/537.36",
                "Phone",
                "Chrome 53.0.2785.124",
                null)
        ));
    }

    @Test
    void testInlineDefinitionDataStream() throws Exception {
        StreamExecutionEnvironment environment = LocalStreamEnvironment.getExecutionEnvironment();

        DataStream<TestRecord> resultDataStream = environment
            .fromElements(
                "Mozilla/5.0 (X11; Linux x86_64) " +
                    "AppleWebKit/537.36 (KHTML, like Gecko) " +
                    "Chrome/48.0.2564.82 Safari/537.36",

                "Mozilla/5.0 (Linux; Android 7.0; Nexus 6 Build/NBD90Z) " +
                    "AppleWebKit/537.36 (KHTML, like Gecko) " +
                    "Chrome/53.0.2785.124 Mobile Safari/537.36"
            )

            .map((MapFunction<String, TestRecord>) TestRecord::new)

            .map(new UserAgentAnalysisMapper<TestRecord>() {
                @Override
                public String getUserAgentString(TestRecord record) {
                    return record.getUserAgent();
                }

                @SuppressWarnings("unused") // Called via the annotation
                @YauaaField("DeviceClass")
                public void setDeviceClass(TestRecord record, String value) {
                    record.deviceClass = value;
                }

                @SuppressWarnings("unused") // Called via the annotation
                @YauaaField("AgentNameVersion")
                public void setAgentNameVersion(TestRecord record, String value) {
                    record.agentNameVersion = value;
                }
            });

        List<TestRecord> result = resultDataStream.executeAndCollect(100);

        assertEquals(2, result.size());

        assertThat(result, hasItems(
            new TestRecord(
                "Mozilla/5.0 (X11; Linux x86_64) " +
                    "AppleWebKit/537.36 (KHTML, like Gecko) " +
                    "Chrome/48.0.2564.82 Safari/537.36",
                "Desktop",
                "Chrome 48.0.2564.82",
                null),

            new TestRecord(
                "Mozilla/5.0 (Linux; Android 7.0; Nexus 6 Build/NBD90Z) " +
                    "AppleWebKit/537.36 (KHTML, like Gecko) " +
                    "Chrome/53.0.2785.124 Mobile Safari/537.36",
                "Phone",
                "Chrome 53.0.2785.124",
                null)
        ));
    }

}
