// Path: redis-storage/src/main/java/com/offbynull/actors/redisclient/SortedSetItem.java
// public final class SortedSetItem {
// 
//     private final double score;
//     private final Object item;
// 
//     /**
//      * Constructs a {@link SortedSetItem} object.
//      * @param score score
//      * @param item item
//      */
//     public SortedSetItem(double score, Object item) {
//         this.score = score;
//         this.item = item;
//     }
// 
//     /**
//      * Get score.
//      * @return score
//      */
//     public double getScore() {
//         return score;
//     }
// 
//     /**
//      * Get item.
//      * @param <T> expected type
//      * @return item
//      * @throws ClassCastException if expected type is not the type stored
//      */
//     public <T> T getItem() {
//         return (T) item;
//     }
// 
// }
// 
// Path: redis-storage/src/main/java/com/offbynull/actors/redisclient/Transaction.java
// public final class Transaction {
// 
//     private final TransactionBlock block;
//     private final boolean retry;
// 
//     /**
//      * Constructs a {@link Transaction} object.
//      * @param retry if {@code true}, transaction will be retried until it passes
//      * @param block block that defines queued transaction operations
//      * @throws NullPointerException if any argument is {@code null}
//      */
//     public Transaction(boolean retry, TransactionBlock block) {
//         Validate.notNull(block);
//         this.retry = retry;
//         this.block = block;
//     }
// 
//     /**
//      * Get retry flag.
//      * @return if {@code true}, transaction will be retried until it passes
//      */
//     public boolean isRetry() {
//         return retry;
//     }
// 
//     /**
//      * Get transaction block.
//      * @return block that defines queued transaction operations
//      */
//     public TransactionBlock getBlock() {
//         return block;
//     }
// 
// }


import com.offbynull.actors.address.Address;
import org.apache.commons.lang3.Validate;
import java.util.Collection;
import com.offbynull.actors.redisclient.Connection;
import com.offbynull.actors.redisclient.ConnectionException;
import com.offbynull.actors.redisclient.SortedSetItem;
import com.offbynull.actors.redisclient.Transaction;
import com.offbynull.actors.redisclient.TransactionResult;
import com.offbynull.actors.redisclient.Watch;
/*
 * Copyright (c) 2017, Kasra Faghihi, All rights reserved.
 * 
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 3.0 of the License, or (at your option) any later version.
 * 
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library.
 */
package com.offbynull.actors.gateways.actor.stores.redis;


/**
 * Timestamp queue -- a queue of (timestamp, address) pairs that's sorted in ascending order by timestamp.
 * <p>
 * Unless there's a critical error, implementations are required to retry indefinity the operation until it succeeds. The following are not
 * considered to be critical errors...
 * <ul>
 * <li>Connection problems.</li>
 * <li>Redis MULTI/EXEC transactions that fail because a WATCH failed are not critical errors.</li>
 * </ul>
 * @author Kasra Faghihi
 */
final class TimestampQueue {
    private static final String KEY_PREFIX = "timestampqueue:";

    private final Connection connection;
    private final String queueKey;

    TimestampQueue(Connection connection, String name, int num) {
        Validate.notNull(name);
        Validate.notNull(connection);

        this.queueKey = KEY_PREFIX + name + ':' + num;
        this.connection = connection;
    }
    
    // remove next item (only if the timestamp for the next item <= currentTimestamp)
    public Address remove(long minTimestamp) throws ConnectionException {
        Validate.isTrue(minTimestamp >= 0L);

        TransactionResult res = connection.transaction(
// Path: redis-storage/src/main/java/com/offbynull/actors/redisclient/SortedSetItem.java
// public final class SortedSetItem {
// 
//     private final double score;
//     private final Object item;
// 
//     /**
//      * Constructs a {@link SortedSetItem} object.
//      * @param score score
//      * @param item item
//      */
//     public SortedSetItem(double score, Object item) {
//         this.score = score;
//         this.item = item;
//     }
// 
//     /**
//      * Get score.
//      * @return score
//      */
//     public double getScore() {
//         return score;
//     }
// 
//     /**
//      * Get item.
//      * @param <T> expected type
//      * @return item
//      * @throws ClassCastException if expected type is not the type stored
//      */
//     public <T> T getItem() {
//         return (T) item;
//     }
// 
// }
// 
// Path: redis-storage/src/main/java/com/offbynull/actors/redisclient/Transaction.java
// public final class Transaction {
// 
//     private final TransactionBlock block;
//     private final boolean retry;
// 
//     /**
//      * Constructs a {@link Transaction} object.
//      * @param retry if {@code true}, transaction will be retried until it passes
//      * @param block block that defines queued transaction operations
//      * @throws NullPointerException if any argument is {@code null}
//      */
//     public Transaction(boolean retry, TransactionBlock block) {
//         Validate.notNull(block);
//         this.retry = retry;
//         this.block = block;
//     }
// 
//     /**
//      * Get retry flag.
//      * @return if {@code true}, transaction will be retried until it passes
//      */
//     public boolean isRetry() {
//         return retry;
//     }
// 
//     /**
//      * Get transaction block.
//      * @return block that defines queued transaction operations
//      */
//     public TransactionBlock getBlock() {
//         return block;
//     }
// 
// }


// Path: redis-storage/src/main/java/com/offbynull/actors/gateways/actor/stores/redis/TimestampQueue.java
import com.offbynull.actors.address.Address;
import org.apache.commons.lang3.Validate;
import java.util.Collection;
import com.offbynull.actors.redisclient.Connection;
import com.offbynull.actors.redisclient.ConnectionException;
import com.offbynull.actors.redisclient.SortedSetItem;
import com.offbynull.actors.redisclient.Transaction;
import com.offbynull.actors.redisclient.TransactionResult;
import com.offbynull.actors.redisclient.Watch;

/*
 * Copyright (c) 2017, Kasra Faghihi, All rights reserved.
 * 
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 3.0 of the License, or (at your option) any later version.
 * 
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library.
 */
package com.offbynull.actors.gateways.actor.stores.redis;


/**
 * Timestamp queue -- a queue of (timestamp, address) pairs that's sorted in ascending order by timestamp.
 * <p>
 * Unless there's a critical error, implementations are required to retry indefinity the operation until it succeeds. The following are not
 * considered to be critical errors...
 * <ul>
 * <li>Connection problems.</li>
 * <li>Redis MULTI/EXEC transactions that fail because a WATCH failed are not critical errors.</li>
 * </ul>
 * @author Kasra Faghihi
 */
final class TimestampQueue {
    private static final String KEY_PREFIX = "timestampqueue:";

    private final Connection connection;
    private final String queueKey;

    TimestampQueue(Connection connection, String name, int num) {
        Validate.notNull(name);
        Validate.notNull(connection);

        this.queueKey = KEY_PREFIX + name + ':' + num;
        this.connection = connection;
    }
    
    // remove next item (only if the timestamp for the next item <= currentTimestamp)
    public Address remove(long minTimestamp) throws ConnectionException {
        Validate.isTrue(minTimestamp >= 0L);

        TransactionResult res = connection.transaction(
                new Transaction(true, queue -> {