<commit_before>package ch.softappeal.yass.core.remote.session;

import ch.softappeal.yass.util.Check;
import ch.softappeal.yass.util.Nullable;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Proxy;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;

public class Reconnector<S extends Session> {

    @FunctionalInterface public interface Connector {
        /**
         * @throws Exception note: will be ignored
         */
        void connect(SessionFactory sessionFactory) throws Exception;
    }

    /**
     * @param executor must interrupt it's threads to terminate reconnects (use {@link ExecutorService#shutdownNow()})
     */
    public final void start(
        final Executor executor, final long initialDelaySeconds, final long delaySeconds,
        final SessionFactory sessionFactory, final Connector connector
    ) {
        Check.notNull(sessionFactory);
        Check.notNull(connector);
        final SessionFactory reconnectorSessionFactory = connection -> {
            final Session session = sessionFactory.create(connection);
            this.session = session;
            return session;
        };
        executor.execute(() -> {
            if (initialDelaySeconds > 0) {
                try {
                    TimeUnit.SECONDS.sleep(initialDelaySeconds);
                } catch (final InterruptedException ignore) {
                    return;
                }
            }
            while (!Thread.interrupted()) {
                final Session session = this.session;
                if ((session == null) || session.isClosed()) {
                    this.session = null;
                    try {
                        connector.connect(reconnectorSessionFactory);
                    } catch (final Exception ignore) {
                        // empty
                    }
                }
                try {
                    TimeUnit.SECONDS.sleep(delaySeconds);
                } catch (final InterruptedException ignore) {
                    return;
                }
            }
        });
    }

    /**
     * @see #start(Executor, long, long, SessionFactory, Connector)
     */
    public final void start(
        final Executor executor, final long delaySeconds,
        final SessionFactory sessionFactory, final Connector connector
    ) {
        start(executor, 0, delaySeconds, sessionFactory, connector);
    }

    private volatile @Nullable Session session = null;

    /**
     * @return current {@link Session}
     * @throws SessionClosedException if no active session
     */
    @SuppressWarnings("unchecked")
    public final S session() throws SessionClosedException {
        final Session session = this.session;
        if (session == null) {
            throw new SessionClosedException();
        }
        return (S)session;
    }

    public final boolean connected() {
        return (session != null);
    }

    @FunctionalInterface public interface SessionProxyGetter<S, C> {
        C get(S session) throws Exception;
    }

    /**
     * @return a proxy surviving reconnect
     */
    @SuppressWarnings("unchecked")
    protected final <C> C proxy(final Class<C> contract, final SessionProxyGetter<S, C> sessionProxyGetter) {
        Check.notNull(sessionProxyGetter);
        return (C)Proxy.newProxyInstance(
            contract.getClassLoader(),
            new Class<?>[] {contract},
            (proxy, method, arguments) -> {
                try {
                    return method.invoke(sessionProxyGetter.get(session()), arguments);
                } catch (final InvocationTargetException e) {
                    throw e.getCause();
                }
            }
        );
    }

}<commit_msg>Reconnector enhanced<commit_after>package ch.softappeal.yass.core.remote.session;

import ch.softappeal.yass.util.Check;
import ch.softappeal.yass.util.Nullable;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Proxy;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;

public class Reconnector<S extends Session> {

    private volatile @Nullable S session = null;

    private static boolean connected(final Session session) {
        return (session != null) && !session.isClosed();
    }

    public final boolean connected() {
        return connected(session);
    }

    /**
     * @return current {@link Session}
     * @throws SessionClosedException if no active session
     */
    public final S session() throws SessionClosedException {
        final S session = this.session;
        if (!connected(session)) {
            throw new SessionClosedException();
        }
        return session;
    }

    @FunctionalInterface public interface SessionProxyGetter<S, C> {
        C get(S session) throws Exception;
    }

    /**
     * @return a proxy surviving reconnect
     */
    @SuppressWarnings("unchecked")
    protected final <C> C proxy(final Class<C> contract, final SessionProxyGetter<S, C> sessionProxyGetter) {
        Check.notNull(sessionProxyGetter);
        return (C)Proxy.newProxyInstance(
            contract.getClassLoader(),
            new Class<?>[] {contract},
            (proxy, method, arguments) -> {
                try {
                    return method.invoke(sessionProxyGetter.get(session()), arguments);
                } catch (final InvocationTargetException e) {
                    throw e.getCause();
                }
            }
        );
    }

    @FunctionalInterface public interface Connector {
        /**
         * @throws Exception note: will be ignored
         */
        void connect(SessionFactory sessionFactory) throws Exception;
    }

    /**
     * @param executor must interrupt it's threads to terminate reconnects (use {@link ExecutorService#shutdownNow()})
     */
    public final void start(
        final Executor executor, final long initialDelaySeconds, final long delaySeconds,
        final SessionFactory sessionFactory, final Connector connector
    ) {
        Check.notNull(sessionFactory);
        Check.notNull(connector);
        @SuppressWarnings("unchecked") final SessionFactory reconnectorSessionFactory = connection -> {
            final Session session = sessionFactory.create(connection);
            this.session = (S)session;
            return session;
        };
        executor.execute(() -> {
            if (initialDelaySeconds > 0) {
                try {
                    TimeUnit.SECONDS.sleep(initialDelaySeconds);
                } catch (final InterruptedException ignore) {
                    return;
                }
            }
            while (!Thread.interrupted()) {
                if (!connected()) {
                    session = null;
                    try {
                        connector.connect(reconnectorSessionFactory);
                    } catch (final Exception ignore) {
                        // empty
                    }
                }
                try {
                    TimeUnit.SECONDS.sleep(delaySeconds);
                } catch (final InterruptedException ignore) {
                    return;
                }
            }
        });
    }

    /**
     * @see #start(Executor, long, long, SessionFactory, Connector)
     */
    public final void start(
        final Executor executor, final long delaySeconds,
        final SessionFactory sessionFactory, final Connector connector
    ) {
        start(executor, 0, delaySeconds, sessionFactory, connector);
    }

}
{'fraction_non_alphanumeric': 0.08969665271966527, 'fraction_numerical': 0.0005230125523012552, 'mean_word_length': 1.9328987730061349, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1891450', 'n_tokens_mistral': 1872, 'n_tokens_neox': 1798, 'n_words': 471}