/*
 * Copyright 2016-2018 Leon Chen
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.moilioncircle.examples.migration;

import com.moilioncircle.examples.migration.cmd.DefaultCommand;
import com.moilioncircle.examples.migration.rdb.MigrationKeyValuePair;
import com.moilioncircle.redis.replicator.CloseListener;
import com.moilioncircle.redis.replicator.Configuration;
import com.moilioncircle.redis.replicator.RedisURI;
import com.moilioncircle.redis.replicator.Replicator;
import com.moilioncircle.redis.replicator.cmd.Command;
import com.moilioncircle.redis.replicator.cmd.CommandListener;
import com.moilioncircle.redis.replicator.rdb.RdbListener;
import com.moilioncircle.redis.replicator.rdb.datatype.KeyValuePair;
import redis.clients.jedis.Client;
import redis.clients.jedis.Protocol;

import java.io.IOException;
import java.net.URISyntaxException;

import static redis.clients.jedis.Protocol.Command.AUTH;
import static redis.clients.jedis.Protocol.Command.RESTORE;

/**
 * @author Leon Chen
 * @since 2.4.3
 */
public class MigrationExample {

    public static void main(String[] args) throws IOException, URISyntaxException {
        sync("redis://127.0.0.1:6379", "redis://127.0.0.1:6380");
    }

    /*
     * Precondition:
     * 1. Make sure the two redis version is same.
     * 2. Make sure the single key-value is not very big.(highly recommend less then 1 MB)
     *
     * We running following steps to sync two redis.
     * 1. Get rdb stream from source redis.
     * 2. Convert source rdb stream to redis dump format.
     * 3. Use Jedis RESTORE command to restore that dump format to target redis.
     * 4. Get aof stream from source redis and sync to target redis.
     */
    @SuppressWarnings("resource")
    public static void sync(String sourceUri, String targetUri) throws IOException, URISyntaxException {
        RedisURI suri = new RedisURI(sourceUri);
        RedisURI turi = new RedisURI(targetUri);
        final ExampleClient target = new ExampleClient(turi.getHost(), turi.getPort());
        Configuration tconfig = Configuration.valueOf(turi);
        if (tconfig.getAuthPassword() != null) {
            String auth = target.send(AUTH, tconfig.getAuthPassword().getBytes());
            System.out.println("AUTH:" + auth);
        }
        Replicator r = new RedisMigrationReplicator(suri.getHost(), suri.getPort(), Configuration.valueOf(suri));
        r.addRdbListener(new RdbListener.Adaptor() {
            @Override
            public void handle(Replicator replicator, KeyValuePair<?> kv) {
                if (!(kv instanceof MigrationKeyValuePair)) return;
                MigrationKeyValuePair mkv = (MigrationKeyValuePair) kv;
                if (mkv.getExpiredMs() == null) {
                    String r = target.restore(mkv.getRawKey(), 0L, mkv.getValue(), true);
                    System.out.println(r);
                } else {
                    long ms = mkv.getExpiredMs() - System.currentTimeMillis();
                    if (ms <= 0) return;
                    String r = target.restore(mkv.getRawKey(), ms, mkv.getValue(), true);
                    System.out.println(r);
                }
            }
        });
        r.addCommandListener(new CommandListener() {
            @Override
            public void handle(Replicator replicator, Command command) {
                if (!(command instanceof DefaultCommand)) return;
                DefaultCommand dc = (DefaultCommand) command;
                String r = target.send(dc.getCommand(), dc.getArgs());
                System.out.println(r);
            }
        });
        r.addCloseListener(new CloseListener() {
            @Override
            public void handle(Replicator replicator) {
                target.close();
            }
        });
        r.open();
    }

    /*
     * Jedis is not a reliable redis client.
     * For simplicity we use Jedis to show this example.
     * In production you need to replace following code to yours.
     * Don't forget to replace DefaultCommandParser too. :)
     */
    public static class ExampleClient extends Client {

        public ExampleClient(final String host, final int port) {
            super(host, port);
        }

        public String send(final Protocol.Command cmd, final byte[]... args) {
            sendCommand(cmd, args);
            return getStatusCodeReply();
        }

        public String restore(byte[] key, long expired, byte[] dumped, boolean replace) {
            if (!replace) {
                return send(RESTORE, key, String.valueOf(expired).getBytes(), dumped);
            } else {
                return send(RESTORE, key, String.valueOf(expired).getBytes(), dumped, "REPLACE".getBytes());
            }
        }
    }
}

@@ -1163,32 +1163,91 @@
 db.RdbListener;%0A
+import com.moilioncircle.redis.replicator.rdb.datatype.DB;%0A
 import com.moili
@@ -1295,24 +1295,24 @@
 yValuePair;%0A
-
 import redis
@@ -1431,24 +1431,74 @@
 axException;
+%0Aimport java.util.concurrent.atomic.AtomicInteger;
 %0A%0Aimport sta
@@ -2898,32 +2898,91 @@
 uth);%0A        %7D%0A
+        final AtomicInteger dbnum = new AtomicInteger(-1);%0A
         Replicat
@@ -3290,32 +3290,359 @@
 ePair)) return;%0A
+                // Step1: select db%0A                DB db = kv.getDb();%0A                int index;%0A                if (db != null && (index = (int) db.getDbNumber()) != dbnum.get()) %7B%0A                    target.select(index);%0A                    dbnum.set(index);%0A                %7D%0A%0A                // Step2: restore dump data%0A
                 
@@ -4346,24 +4346,24 @@
  command) %7B%0A
-
             
@@ -4408,32 +4408,75 @@
 mmand)) return;%0A
+                // Step3: sync aof command%0A
                 

fix example