package de.serosystems.grx.example;

import de.serosystems.proto.v3.grx.receiverd.ReceiverDProto;
import de.serosystems.proto.v3.grx.receiverd.ReceiverdGrpc;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * Example: Subscribe to Mode S downlink frames asynchronously using a StreamObserver.
 *
 * <p>This example connects to the Receiverd service and opens a server-streaming RPC
 * to receive Mode S downlink frames for DF 11 and DF 17. A {@link StreamObserver}
 * callback handles incoming frames on a background thread. The main thread waits
 * for 5 seconds, then shuts down.</p>
 *
 * <p>Usage: {@code java AsyncModeSDownlink [host]}</p>
 * <p>Default host: YOUR_GRX_IP</p>
 */
public class AsyncModeSDownlink {

	/** Default IP address of the GRX receiver. */
	private static final String DEFAULT_HOST = "YOUR_GRX_IP";

	/** Receiverd gRPC port. */
	private static final int RECEIVERD_PORT = 5303;

	/** How long to run the subscription [seconds]. */
	private static final int RUN_DURATION_SECONDS = 5;

	/**
	 * A CountDownLatch is used to block the main thread while the async stream
	 * runs in the background. It counts down when the stream completes or errors.
	 */
	private static final CountDownLatch finishLatch = new CountDownLatch(1);

	/**
	 * StreamObserver callback for incoming Mode S downlink frames.
	 * Each received frame is printed to stdout.
	 */
	private static class ModeSDownlinkFrameHandler
			implements StreamObserver<ReceiverDProto.ModeSDownlinkFrameWithStreamInfo> {

		@Override
		public void onNext(ReceiverDProto.ModeSDownlinkFrameWithStreamInfo msg) {
			ReceiverDProto.ModeSDownlinkFrame frame = msg.getFrame();

			// Convert payload bytes to hex string
			StringBuilder hex = new StringBuilder();
			for (byte b : frame.getPayload().toByteArray()) {
				hex.append(String.format("%02x", b));
			}

			System.out.printf("Message: t=%d  signal=%.1f dBm  payload=%s%n",
					frame.getTimestamp(),
					frame.getLevelSignal(),
					hex);
		}

		@Override
		public void onError(Throwable throwable) {
			System.err.printf("Stream error: %s%n", throwable.getMessage());
			// Signal the main thread that we are done
			finishLatch.countDown();
		}

		@Override
		public void onCompleted() {
			System.out.println("Stream completed.");
			// Signal the main thread that we are done
			finishLatch.countDown();
		}
	}

	public static void main(String[] args) {
		// Use the first CLI argument as hostname, or fall back to the default
		String host = args.length > 0 ? args[0] : DEFAULT_HOST;

		// Create a plaintext gRPC channel to the Receiverd service
		ManagedChannel channel = ManagedChannelBuilder.forAddress(host, RECEIVERD_PORT)
				.usePlaintext()
				.build();

		try {
			// Create an async stub for the Receiverd service
			ReceiverdGrpc.ReceiverdStub asyncStub = ReceiverdGrpc.newStub(channel);

			// Build the subscription request for DF 11 and DF 17
			ReceiverDProto.GetModeSDownlinkFramesRequest request =
					ReceiverDProto.GetModeSDownlinkFramesRequest.newBuilder()
							.addDownlinkFormats(11)  // All-Call Reply
							.addDownlinkFormats(17)  // Extended Squitter (ADS-B)
							.build();

			// Start the async streaming call with our handler
			asyncStub.getModeSDownlinkFrames(request, new ModeSDownlinkFrameHandler());

			System.out.printf("Subscribed to DF 11 + DF 17. Listening for %d seconds...%n",
					RUN_DURATION_SECONDS);

			// Wait for the specified duration or until the stream ends/errors
			finishLatch.await(RUN_DURATION_SECONDS, TimeUnit.SECONDS);

		} catch (InterruptedException e) {
			System.err.println("Interrupted: " + e.getMessage());
		} finally {
			// Shut down the channel, which also cancels any active streams
			channel.shutdownNow();
		}
	}
}