package de.serosystems.proto.v3.backend.api;

import io.grpc.*;
import io.grpc.stub.StreamObserver;

import java.io.File;
import java.io.IOException;
import java.util.HashMap;
import java.util.UUID;

import static de.serosystems.proto.v3.backend.api.BlockingExample.printSensorInfo;
import static de.serosystems.proto.v3.backend.api.BlockingExample.printTargetReports;

/**
 * This example shows how to use the SecureTrack backend API in
 * a non-blocking (asynchronous) fashion.
 *
 * @author Matthias Schäfer (schaefer@sero-systems.de)
 */
public class AsynchronousExample {

    private static final String API_ADDRESS = "api.secureadsb.com";
    private static final int API_PORT = 4201;
    private static final UUID token = UUID.fromString("INSERT YOUR TOKEN HERE");
    private static final String TLS_CA = "ca.crt"; // point this to your ca.crt

    private static class SensorInfoSink implements StreamObserver<SeRoAPIProto.SensorInfoResponse> {
        boolean finished = false;

        @Override
        public void onNext(SeRoAPIProto.SensorInfoResponse response) {
            // called when data is returned
            printSensorInfo(response.getSensorInfoList());
        }

        @Override
        public void onError(Throwable t) {
            // called when an error occurs
            finished = true;
            System.err.println("Retrieving sensor information failed!");
            System.err.println("Reason: " + t.getMessage());
        }

        @Override
        public void onCompleted() {
            // called when the request is completed
            finished = true;
            System.out.println("Retrieving sensor data completed!");
        }

        public boolean isRunning() {
            return !finished;
        }
    }

    private static class TargetReportsSink implements StreamObserver<SeRoAPIProto.TargetReport> {
        boolean finished = false;
        HashMap<Integer, SeRoAPIProto.TargetReport> targetReports = new HashMap<>();
        long count = 0L;

        @Override
        public void onNext(SeRoAPIProto.TargetReport report) {
            // called when new data arrives
            count++;
            targetReports.put(report.getTarget().getAddress(), report);
        }

        @Override
        public void onError(Throwable t) {
            // called when an error occurs
            synchronized (this) {
                finished = true;
                if (Status.fromThrowable(t).getCode() == Status.Code.CANCELLED) {
                    System.out.println("Target report stream was cancelled.");
                } else {
                    System.err.println("An error occurred while retrieving target reports or stream was closed!");
                    System.err.println("Reason: " + t.getMessage());
                }
                this.notifyAll(); // wake up whoever is waiting
            }
        }

        @Override
        public void onCompleted() {
            // Note: this should only happen if client closes the stream or the server is restarted
            synchronized (this) {
                finished = true;
                System.out.println("Stream was closed!");
                this.notifyAll(); // wake up whoever is waiting
            }
        }

        /**
         * Blocks until the stream is closed or timeout fires
         * @param timeout in milliseconds (0 disables timeout)
         * @return true if stream finished
         */
        public boolean waitUntilFinished(long timeout) throws InterruptedException {
            synchronized (this) {
                if (finished) return true;
                this.wait(timeout); // wait
            }

            return finished;
        }
    }

    public static void main(String[] args) {

        System.out.println("Hello world! This is an example for using the SecureTrack API with asynchronous gRPC calls!");

        System.out.printf("Connecting to SeRo API at %s:%d.\n", API_ADDRESS, API_PORT);

        // first create channel
        final ChannelCredentials credentials;
        try {
            credentials = TlsChannelCredentials.newBuilder()
                    .trustManager(new File(TLS_CA))
                    .build();
        } catch (IOException e) {
            System.err.println("Could not load TLS certificate: " + e.getMessage());
            return;
        }
        ManagedChannel ch = Grpc.newChannelBuilderForAddress(API_ADDRESS, API_PORT, credentials)
                .build();

        // then create stub
        SeRoAPIGrpc.SeRoAPIStub stub = SeRoAPIGrpc.newStub(ch);

        System.out.println("Retrieving info about all sensors that are available to this token.");

        // prepare request
        SeRoAPIProto.SensorInfoRequest sensorRequest = SeRoAPIProto.SensorInfoRequest.newBuilder()
                .setToken(token.toString())
                .build();

        SensorInfoSink sensorInfoObserver = new SensorInfoSink();

        // send request and return immediately
        stub.getSensorInfo(sensorRequest, sensorInfoObserver);

        System.out.println("Sleeping in the meantime...");
        try {
            while (sensorInfoObserver.isRunning())
                Thread.sleep(5_000L);
        } catch (InterruptedException e) {
            System.err.println("Something woke me up unexpectedly: " + e.getMessage());
        }

        System.out.println("Ok, that nap was great! Now on to other things...");

        System.out.println("Let's see what happens when we request info for a sensor that does not belong to us!");

        sensorRequest = SeRoAPIProto.SensorInfoRequest.newBuilder()
                .setToken(token.toString())
                .addSensors(SeRoAPIProto.Sensor.newBuilder()
                        .setSerial(12345L)
                        .setType(SeRoAPIProto.Sensor.Type.GRX1090)
                        .build())
                .build();

        sensorInfoObserver = new SensorInfoSink();
        stub.getSensorInfo(sensorRequest, sensorInfoObserver);

        // wait for call to finish
        try {
            while (sensorInfoObserver.isRunning())
                Thread.sleep(100L);
        } catch (InterruptedException e) {
            System.err.println("Something woke me up unexpectedly: " + e.getMessage());
        }

        System.out.println("Now we are going to retrieve target state reports for about 10 seconds.");

        SeRoAPIProto.TargetReportsRequest targetReportsRequest = SeRoAPIProto.TargetReportsRequest.newBuilder()
                .setToken(token.toString())
                .build();

        TargetReportsSink targetReportsSink = new TargetReportsSink();

        // since we want to cancel it after 10 seconds, we need to do the call within a cancellable context
        Context.CancellableContext withCancellation = Context.current().withCancellation();
        Context ctx = withCancellation.attach();
        try {
            stub.getTargetReports(targetReportsRequest, targetReportsSink);
            Thread.sleep(10_000);
        } catch (InterruptedException e) {
            System.err.println("Someone woke me up early :-(");
        } finally {
            withCancellation.close();
            withCancellation.detach(ctx);
        }

        // wait for stream to finish gracefully
        try {
            targetReportsSink.waitUntilFinished(0L);
        } catch (InterruptedException e) {
            System.err.println("Something interrupted me while I was waiting for the stream to finish :-(");
        }

        System.out.printf("Done! We received %d target reports from %d different aircraft!\n",
                targetReportsSink.count, targetReportsSink.targetReports.size());

        printTargetReports(targetReportsSink.targetReports.values());
    }

}