/*
 * 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.Protocol;

import java.io.IOException;

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 {
        sync("127.0.0.1", 6379, "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 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.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());
            }
        }
    }
}

@@ -888,16 +888,68 @@
 ration;%0A
+import com.moilioncircle.redis.replicator.RedisURI;%0A
 import c
@@ -1337,24 +1337,117 @@
 OException;%0A
+import java.net.URISyntaxException;%0A%0Aimport static redis.clients.jedis.Protocol.Command.AUTH;
 %0Aimport stat
@@ -1627,32 +1627,52 @@
 rows IOException
+, URISyntaxException
  %7B%0A        sync(
@@ -1672,16 +1672,24 @@
   sync(%22
+redis://
 127.0.0.
@@ -1693,18 +1693,25 @@
 .0.1
-%22, 
+:
 6379
+%22
 , %22
+redis://
 127.
@@ -1719,15 +1719,14 @@
 .0.1
-%22, 
+:
 6380
+%22
 );%0A 
@@ -2300,85 +2300,169 @@
 urce
-Host, int sourcePort, String targetHost, int targetPort) throws IOException %7B
+Uri, String targetUri) throws IOException, URISyntaxException %7B%0A        RedisURI suri = new RedisURI(sourceUri);%0A        RedisURI turi = new RedisURI(targetUri);
 %0A   
@@ -2518,31 +2518,290 @@
 nt(t
-ar
+uri.
 getHost
+()
 , t
-argetPort);
+uri.getPort());%0A        Configuration tconfig = Configuration.valueOf(turi);%0A        if (tconfig.getAuthPassword() != null) %7B%0A            String auth = target.send(AUTH, tconfig.getAuthPassword().getBytes());%0A            System.out.println(%22AUTH:%22 + auth);%0A        %7D
 %0A   
@@ -2854,29 +2854,37 @@
 or(s
-o
 ur
-ce
+i.get
 Host
+()
 , s
-o
 ur
-ce
+i.get
 Port
+()
 , Co
@@ -2899,23 +2899,20 @@
 ion.
-defaultSetting(
+valueOf(suri
 ));%0A

fix example