/*
 * 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.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.Connection;
import redis.clients.jedis.Protocol;

import java.io.IOException;

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

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

    @SuppressWarnings("resource")
    public static void sync(String sourceHost, int sourcePort, String targetHost, int targetPort) throws IOException {
        final ExampleClient target = new ExampleClient(targetHost, targetPort);
        Replicator r = new RedisMigrationReplicator(sourceHost, sourcePort, Configuration.defaultSetting());
        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.send(Protocol.Command.RESTORE, mkv.getRawKey(), "0".getBytes(), mkv.getValue(), "REPLACE".getBytes());
                    System.out.println(r);
                } else {
                    long ms = mkv.getExpiredMs() - System.currentTimeMillis();
                    if (ms <= 0) return;
                    String r = target.send(Protocol.Command.RESTORE, mkv.getRawKey(), String.valueOf(ms).getBytes(), mkv.getValue(), "REPLACE".getBytes());
                    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();
    }

    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 Connection sendCommand(final Protocol.Command cmd, final byte[]... args) {
            return super.sendCommand(cmd, args);
        }
    }
}

@@ -1254,27 +1254,62 @@
 dis.
-Connec
+Protocol;%0A%0Aimport java.io.IOExcep
 tion;%0A
+%0A
 import 
+static 
 redi
@@ -1336,37 +1336,24 @@
 ocol
-;%0A%0Aimport java.io.IOException
+.Command.RESTORE
 ;%0A%0A/
@@ -1548,24 +1548,514 @@
 80);%0A    %7D%0A%0A
+    /*%0A     * Precondition:%0A     * 1. Make sure the two redis version is same.%0A     * 2. Make sure the single key-value is not very big.(highly recommend less then 1 MB)%0A     *%0A     * We running following steps to sync two redis.%0A     * 1. Get rdb stream from source redis.%0A     * 2. Convert source rdb stream to redis dump format.%0A     * 3. Use Jedis RESTORE command to restore that dump format to target redis.%0A     * 4. Get aof stream from source redis and sync to target redis.%0A     */%0A
     @Suppres
@@ -2767,39 +2767,16 @@
 get.
-send(Protocol.Command.RESTORE, 
+restore(
 mkv.
@@ -2792,22 +2792,10 @@
 (), 
-%220%22.getBytes()
+0L
 , mk
@@ -2808,36 +2808,20 @@
 alue(), 
-%22REPLACE%22.getBytes()
+true
 );%0A     
@@ -3045,123 +3045,57 @@
 get.
-send(Protocol.Command.RESTORE, mkv.getRawKey(), String.valueOf(ms).getBytes(), mkv.getValue(), %22REPLACE%22.getBytes()
+restore(mkv.getRawKey(), ms, mkv.getValue(), true
 );%0A 
@@ -3798,24 +3798,267 @@
 n();%0A    %7D%0A%0A
+    /*%0A     * Jedis is not a reliable redis client.%0A     * For simplicity we use Jedis to show this example.%0A     * In production you need to replace following code to yours.%0A     * Don't forget to replace DefaultCommandParser too. :)%0A     */%0A
     public s
@@ -4395,131 +4395,341 @@
 lic 
-Connection sendCommand(final Protocol.Command cmd, final byte%5B%5D... args) %7B%0A            return super.sendCommand(cmd, args);
+String restore(byte%5B%5D key, long expired, byte%5B%5D dumped, boolean replace) %7B%0A            if (!replace) %7B%0A                return send(RESTORE, key, String.valueOf(expired).getBytes(), dumped);%0A            %7D else %7B%0A                return send(RESTORE, key, String.valueOf(expired).getBytes(), dumped, %22REPLACE%22.getBytes());%0A            %7D
 %0A   

fix example