package study.gateway;

import java.net.*;
import java.io.*;
import java.util.Set;
import java.util.concurrent.*;

/** Owned loopback TCP link. Faults close only this lab's sockets, never the Redis service. */
public final class RedisLink implements AutoCloseable {
    private final ServerSocket listener;
    private final ExecutorService workers=Executors.newVirtualThreadPerTaskExecutor();
    private final Set<Socket> sockets=ConcurrentHashMap.newKeySet();
    private volatile boolean failed;
    private volatile boolean closed;
    private final String host;
    private final int redisPort;
    RedisLink(String host,int redisPort) throws IOException {
        this.host=host; this.redisPort=redisPort;
        listener=new ServerSocket(); listener.bind(new InetSocketAddress("127.0.0.1",0));
        workers.submit(() -> {
            while(!closed) {
                try {
                    Socket client=listener.accept(); sockets.add(client);
                    if(failed) { client.close(); sockets.remove(client); continue; }
                    Socket upstream=new Socket(); upstream.connect(new InetSocketAddress(host,redisPort),500);
                    sockets.add(upstream);
                    workers.submit(() -> relay(client,upstream));
                    workers.submit(() -> relay(upstream,client));
                } catch(IOException e) { if(!closed&&!failed) System.out.println("U32 redisLinkConnectionFailure=true"); }
            }
        });
    }
    private void relay(Socket source,Socket target) {
        try { source.getInputStream().transferTo(target.getOutputStream()); }
        catch(IOException ignored) { }
        finally { closeSocket(source); closeSocket(target); }
    }
    private void closeSocket(Socket socket) {
        try { socket.close(); } catch(IOException ignored) { }
        sockets.remove(socket);
    }
    int port() { return listener.getLocalPort(); }
    void fail(boolean value) { failed=value; if(value)sockets.forEach(this::closeSocket); }
    public void close() throws IOException {
        closed=true; listener.close(); sockets.forEach(this::closeSocket); workers.close();
        System.out.println("U32 redisLinkClosed=true");
    }
}