/*******************************************************************************
 * Copyright (C) 2023 DiffusionData Ltd.
 *
 * 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 com.pushtechnology.diffusion.manual;

import static com.pushtechnology.diffusion.client.Diffusion.dataTypes;
import static com.pushtechnology.diffusion.client.Diffusion.newTopicSpecification;

import com.pushtechnology.diffusion.client.Diffusion;
import com.pushtechnology.diffusion.client.callbacks.ErrorReason;
import com.pushtechnology.diffusion.client.features.TopicUpdate;
import com.pushtechnology.diffusion.client.features.Topics;
import com.pushtechnology.diffusion.client.features.UpdateConstraint;
import com.pushtechnology.diffusion.client.session.Session;
import com.pushtechnology.diffusion.client.topics.details.TopicSpecification;
import com.pushtechnology.diffusion.client.topics.details.TopicType;
import com.pushtechnology.diffusion.datatype.json.JSON;
import com.pushtechnology.diffusion.datatype.json.JSONDataType;

import java.util.concurrent.CompletionException;

/**
 * An example of using a client to update a JSON topic
 *
 * This uses the 'TopicUpdate' feature.
 *
 * @author DiffusionData Limited
 */
public final class TopicUpdateJSONExample {

    public static void main(String[] args) {

        final Session session = Diffusion.sessions()
            .principal("admin")
            .password("password")
            .open("ws://localhost:8080");

        final TopicUpdate topicUpdate = session.feature(TopicUpdate.class);
        final JSONDataType jsonDataType = dataTypes().json();
        final String path = "my/topic/path";

        // add a value stream and subscribe to receive topic updates
        session.feature(Topics.class)
            .addStream(">my/topic/path", JSON.class, new MyValueStream());
        session.feature(Topics.class).subscribe(">my/topic/path");

        // add a new JSON topic and set an initial value
        topicUpdate.addAndSet(
            path,
            newTopicSpecification(TopicType.JSON),
            JSON.class,
            jsonDataType.fromJsonString("{ \"diffusion\": \"data\" }")
        ).join();

        // set a new value for our JSON topic
        topicUpdate.set(
            path,
            JSON.class,
            jsonDataType.fromJsonString("{ \"hello\": \"world\" }")
        ).join();

        // update the topic only if the JSON value of our topic matches
        // our constraint value, we expect this to fail
        final UpdateConstraint.Factory constraints = Diffusion.updateConstraints();

        try {
            topicUpdate.set(
                path,
                JSON.class,
                jsonDataType.fromJsonString("{ \"new\": \"data\" }"),
                constraints.jsonValue().with("/hello", String.class, "darkness")
            ).join();
        }
        catch (CompletionException e) {
            System.out.println("Expected failure: " + e.getCause());
        }

        // apply a JSON patch to replace the value of our JSON data
        topicUpdate.applyJsonPatch(
            path,
            "[{\"op\": \"replace\", \"path\": \"/hello\", \"value\":\"there\" }]"
        ).join();

        session.close();
    }

    private static class MyValueStream implements Topics.ValueStream<JSON> {
        @Override
        public void onValue(String topicPath,
            TopicSpecification specification, JSON oldValue, JSON newValue) {
            System.out.printf("%s updated : %s\n", topicPath, newValue.toJsonString());
        }

        @Override
        public void onSubscription(String topicPath,
            TopicSpecification specification) { }

        @Override
        public void onUnsubscription(String topicPath,
            TopicSpecification specification, Topics.UnsubscribeReason reason) { }

        @Override
        public void onClose() { }

        @Override
        public void onError(ErrorReason errorReason) { }
    }
}