package be.bagofwords.db.remote;

import be.bagofwords.application.memory.MemoryManager;
import be.bagofwords.cache.CachesManager;
import be.bagofwords.db.DataInterface;
import be.bagofwords.db.DataInterfaceFactory;
import be.bagofwords.db.combinator.Combinator;
import be.bagofwords.ui.UI;
import be.bagofwords.util.SafeThread;
import be.bagofwords.util.WrappedSocketConnection;
import org.apache.commons.io.IOUtils;

import java.io.IOException;
import java.net.SocketException;
import java.util.HashMap;
import java.util.Map;

import static be.bagofwords.application.BaseServer.LONG_OK;

public class RemoteDatabaseInterfaceFactory extends DataInterfaceFactory {

    private final String host;
    private final int port;
    private Map<String, DataInterface> dataInterfaceMap;
    private ChangedValueListenerThread changedValueListenerThread;

    public RemoteDatabaseInterfaceFactory(CachesManager cachesManager, MemoryManager memoryManager, String host, int port) {
        super(cachesManager, memoryManager);
        this.host = host;
        this.port = port;
        this.dataInterfaceMap = new HashMap<>();
    }

    @Override
    protected synchronized <T extends Object> DataInterface<T> createBaseDataInterface(String nameOfSubset, Class<T> objectClass, Combinator<T> combinator) {
        if (changedValueListenerThread == null) {
            synchronized (this) {
                try {
                    this.changedValueListenerThread = new ChangedValueListenerThread();
                    this.changedValueListenerThread.start();
                } catch (IOException e) {
                    throw new RuntimeException("Unexpected exception while starting changedValueListenerThread", e);
                }
            }
        }
        DataInterface result = new RemoteDataInterface<>(nameOfSubset, objectClass, combinator, host, port);
        dataInterfaceMap.put(nameOfSubset, result);
        return result;
    }

    @Override
    public synchronized void close() {
        if (changedValueListenerThread != null) {
            changedValueListenerThread.terminateAndWait();
        }
        super.close();
    }

    private class ChangedValueListenerThread extends SafeThread {

        private WrappedSocketConnection connection;

        public ChangedValueListenerThread() throws IOException {
            super("ChangedValueListener", false);
            connection = new WrappedSocketConnection(host, port);
        }

        @Override
        protected void runInt() throws Exception {
            connection.writeByte((byte) RemoteDataInterfaceServer.Action.LISTEN_TO_CHANGES.ordinal());
            connection.flush();
            try {
                while (!isTerminateRequested()) {
                    String interfaceName = connection.readString();
                    int numOfKeys = connection.readInt();
                    long[] keys = new long[numOfKeys];
                    for (int i = 0; i < numOfKeys; i++) {
                        keys[i] = connection.readLong();
                    }
                    DataInterface dataInterface = dataInterfaceMap.get(interfaceName);
                    if (dataInterface != null) {
                        dataInterface.notifyListenersOfChangedValues(keys);
                    }
                    connection.writeLong(LONG_OK);
                    connection.flush();
                }
            } catch (SocketException exp) {
                if (!exp.getMessage().equals("Socket closed")) {
                    UI.writeError("Error in ChangedValueListener", exp);
                }
            }
            IOUtils.closeQuietly(connection);
        }

        @Override
        protected void doTerminate() {
            IOUtils.closeQuietly(connection);
        }
    }
}

@@ -209,16 +209,100 @@
 actory;%0A
+import be.bagofwords.db.application.environment.RemoteCountDBEnvironmentProperties;%0A
 import b
@@ -340,16 +340,16 @@
 inator;%0A
-
 import b
@@ -494,16 +494,79 @@
 IOUtils;
+%0Aimport org.springframework.beans.factory.annotation.Autowired;
 %0A%0Aimport
@@ -922,16 +922,16 @@
 aceMap;%0A
-
     priv
@@ -990,16 +990,344 @@
 hread;%0A%0A
+    @Autowired%0A    public RemoteDatabaseInterfaceFactory(CachesManager cachesManager, MemoryManager memoryManager, RemoteCountDBEnvironmentProperties environmentProperties) %7B%0A        this(cachesManager, memoryManager, environmentProperties.getDatabaseServerAddress(), environmentProperties.getDataInterfaceServerPort());%0A    %7D%0A%0A
     publ

Make sure RemoteDatabaseInterfaceFactory can be wired in spring