// Path: src/main/java/biz/paluch/spinach/api/DisqueConnection.java
// public interface DisqueConnection<K, V> extends StatefulConnection<K, V> {
// 
//     /**
//      * Returns the {@link DisqueCommands} API for the current connection. Does not create a new connection.
//      * 
//      * @return the synchronous API for the underlying connection.
//      */
//     DisqueCommands<K, V> sync();
// 
//     /**
//      * Returns the {@link DisqueAsyncCommands} API for the current connection. Does not create a new connection.
//      * 
//      * @return the asynchronous API for the underlying connection.
//      */
//     DisqueAsyncCommands<K, V> async();
// 
//     /**
//      * Returns the {@link DisqueReactiveCommands} API for the current connection. Does not create a new connection.
//      *
//      * @return the reactive API for the underlying connection.
//      */
//     DisqueReactiveCommands<K, V> reactive();
// 
// }
// 
// Path: src/main/java/biz/paluch/spinach/api/Job.java
// public class Job<K, V> {
//     private K queue;
//     private String id;
//     private V body;
//     private Map<String, Long> counters;
// 
//     protected Job() {
//     }
// 
//     public Job(K queue, String id, V body, Map<String, Long> counters) {
//         this.queue = queue;
//         this.id = id;
//         this.body = body;
//         this.counters = counters;
//     }
// 
//     /**
//      *
//      * @return the queue
//      */
//     public K getQueue() {
//         return queue;
//     }
// 
//     /**
//      *
//      * @return the JobId
//      */
//     public String getId() {
//         return id;
//     }
// 
//     /**
//      *
//      * @return the Job body
//      */
//     public V getBody() {
//         return body;
//     }
// 
//     /**
//      * If requested with a WITHCOUNTERS flag, getjob also populates a counters field.
//      *
//      * @return map of counters
//      */
//     public Map<String, Long> getCounters() { return counters; }
// }


import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;
import biz.paluch.spinach.api.DisqueConnection;
import biz.paluch.spinach.api.Job;
import io.netty.util.internal.ConcurrentSet;
import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;
import rx.Observable;
import rx.Scheduler;
import rx.Subscriber;
import rx.Subscription;
import rx.functions.Action0;

        if (reconnectTrigger != null) {
            reconnectTrigger.unsubscribe();
            reconnectTrigger = null;
        }
    }

    /**
     * Enable job locality tracking.
     */
    void withJobLocalityTracking() {
        this.jobLocalityTracking = true;
    }

    void withNodeSwitching(long nodeReconnectCheckInterval, TimeUnit nodeReconnectCheckTimeUnit) {
        this.improveLocalityInterval = nodeReconnectCheckInterval;
        this.improveLocalityTimeUnit = nodeReconnectCheckTimeUnit;
    }

    /**
     * Initiate the switch nodes check.
     */
    void switchNodes() {
        for (GetJobsAction action : actions) {
            action.switchNodes();
        }
    }

    static class LocalityAwareConnection<K, V> {
        private final NodeIdAwareSocketAddressSupplier socketAddressSupplier;
// Path: src/main/java/biz/paluch/spinach/api/DisqueConnection.java
// public interface DisqueConnection<K, V> extends StatefulConnection<K, V> {
// 
//     /**
//      * Returns the {@link DisqueCommands} API for the current connection. Does not create a new connection.
//      * 
//      * @return the synchronous API for the underlying connection.
//      */
//     DisqueCommands<K, V> sync();
// 
//     /**
//      * Returns the {@link DisqueAsyncCommands} API for the current connection. Does not create a new connection.
//      * 
//      * @return the asynchronous API for the underlying connection.
//      */
//     DisqueAsyncCommands<K, V> async();
// 
//     /**
//      * Returns the {@link DisqueReactiveCommands} API for the current connection. Does not create a new connection.
//      *
//      * @return the reactive API for the underlying connection.
//      */
//     DisqueReactiveCommands<K, V> reactive();
// 
// }
// 
// Path: src/main/java/biz/paluch/spinach/api/Job.java
// public class Job<K, V> {
//     private K queue;
//     private String id;
//     private V body;
//     private Map<String, Long> counters;
// 
//     protected Job() {
//     }
// 
//     public Job(K queue, String id, V body, Map<String, Long> counters) {
//         this.queue = queue;
//         this.id = id;
//         this.body = body;
//         this.counters = counters;
//     }
// 
//     /**
//      *
//      * @return the queue
//      */
//     public K getQueue() {
//         return queue;
//     }
// 
//     /**
//      *
//      * @return the JobId
//      */
//     public String getId() {
//         return id;
//     }
// 
//     /**
//      *
//      * @return the Job body
//      */
//     public V getBody() {
//         return body;
//     }
// 
//     /**
//      * If requested with a WITHCOUNTERS flag, getjob also populates a counters field.
//      *
//      * @return map of counters
//      */
//     public Map<String, Long> getCounters() { return counters; }
// }


// Path: src/main/java/biz/paluch/spinach/cluster/QueueListener.java
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;
import biz.paluch.spinach.api.DisqueConnection;
import biz.paluch.spinach.api.Job;
import io.netty.util.internal.ConcurrentSet;
import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;
import rx.Observable;
import rx.Scheduler;
import rx.Subscriber;
import rx.Subscription;
import rx.functions.Action0;


        if (reconnectTrigger != null) {
            reconnectTrigger.unsubscribe();
            reconnectTrigger = null;
        }
    }

    /**
     * Enable job locality tracking.
     */
    void withJobLocalityTracking() {
        this.jobLocalityTracking = true;
    }

    void withNodeSwitching(long nodeReconnectCheckInterval, TimeUnit nodeReconnectCheckTimeUnit) {
        this.improveLocalityInterval = nodeReconnectCheckInterval;
        this.improveLocalityTimeUnit = nodeReconnectCheckTimeUnit;
    }

    /**
     * Initiate the switch nodes check.
     */
    void switchNodes() {
        for (GetJobsAction action : actions) {
            action.switchNodes();
        }
    }

    static class LocalityAwareConnection<K, V> {
        private final NodeIdAwareSocketAddressSupplier socketAddressSupplier;
        private final DisqueConnection<K, V> connection;