/*
 * 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 {
    @SuppressWarnings("resource")
    public static void main(String[] args) throws IOException {
        final MyClient target = new MyClient("127.0.0.1", 6380);
        Replicator r = new RedisMigrationReplicator("127.0.0.1", 6379, 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) {
                    target.sendCommand(Protocol.Command.RESTORE, mkv.getRawKey(), "0".getBytes(), mkv.getValue(), "REPLACE".getBytes());
                    String r = target.getStatusCodeReply();
                    System.out.println(r);
                } else {
                    long ms = mkv.getExpiredMs() - System.currentTimeMillis();
                    if (ms > 0) {
                        target.sendCommand(Protocol.Command.RESTORE, mkv.getRawKey(), String.valueOf(ms).getBytes(), mkv.getValue(), "REPLACE".getBytes());
                        String r = target.getStatusCodeReply();
                        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;
                target.sendCommand(dc.getCommand(), dc.getArgs());
                String r = target.getStatusCodeReply();
                System.out.println(r);
            }
        });
        r.addCloseListener(new CloseListener() {
            @Override
            public void handle(Replicator replicator) {
                target.close();
            }
        });
        r.open();
    }

    public static class MyClient extends Client {
        public MyClient(final String host, final int port) {
            super(host, port);
        }

        public Connection sendCommand(final Protocol.Command cmd, final byte[]... args) {
            return super.sendCommand(cmd, args);
        }
    }
}

@@ -1410,41 +1410,8 @@
 e %7B%0A
-    @SuppressWarnings(%22resource%22)
 %0A   
@@ -1483,62 +1483,289 @@
     
-final MyClient target = new MyClient(%22127.0.0.1%22, 6380
+sync(%22127.0.0.1%22, 6379, %22127.0.0.1%22, 6380);%0A    %7D%0A%0A    @SuppressWarnings(%22resource%22)%0A    public static void sync(String sourceHost, int sourcePort, String targetHost, int targetPort) throws IOException %7B%0A        final ExampleClient target = new ExampleClient(targetHost, targetPort
 );%0A 
@@ -1815,33 +1815,38 @@
 licator(
-%22127.0.0.1%22, 6379
+sourceHost, sourcePort
 , Config
@@ -2224,32 +2224,43 @@
                 
+ String r =
  target.sendComm
@@ -2247,39 +2247,32 @@
  r = target.send
-Command
 (Protocol.Comman
@@ -2358,68 +2358,8 @@
 ));%0A
-                    String r = target.getStatusCodeReply();%0A
     
@@ -2532,14 +2532,21 @@
 (ms 
-%3E
+%3C=
  0) 
-%7B
+return;
 %0A   
@@ -2554,35 +2554,42 @@
                 
-   
+String r =
  target.sendComm
@@ -2584,23 +2584,16 @@
 get.send
-Command
 (Protoco
@@ -2722,120 +2722,30 @@
     
-    String r = target.getStatusCodeReply();%0A                        System.out.println(r);%0A                    %7D
+System.out.println(r);
 %0A   
@@ -3068,32 +3068,43 @@
 %0A               
+ String r =
  target.sendComm
@@ -3099,23 +3099,16 @@
 get.send
-Command
 (dc.getC
@@ -3136,64 +3136,8 @@
 ));%0A
-                String r = target.getStatusCodeReply();%0A
     
@@ -3431,18 +3431,23 @@
 c class 
-My
+Example
 Client e
@@ -3462,16 +3462,17 @@
 lient %7B%0A
+%0A
         
@@ -3482,10 +3482,15 @@
 lic 
-My
+Example
 Clie
@@ -3567,24 +3567,191 @@
 %0A        %7D%0A%0A
+        public String send(final Protocol.Command cmd, final byte%5B%5D... args) %7B%0A            sendCommand(cmd, args);%0A            return getStatusCodeReply();%0A        %7D%0A%0A
         publ

fix example