/*-
 * See the file LICENSE for redistribution information.
 * 
 * Copyright (c) 2010, 2013 Oracle and/or its affiliates.  All rights reserved.
 *
 */

package repmgrtests;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;

import java.io.BufferedReader;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.net.Socket;

import org.junit.Before;
import org.junit.Test;

import com.sleepycat.db.BtreeStats;
import com.sleepycat.db.Database;
import com.sleepycat.db.DatabaseConfig;
import com.sleepycat.db.DatabaseEntry;
import com.sleepycat.db.DatabaseType;
import com.sleepycat.db.Environment;
import com.sleepycat.db.EnvironmentConfig;
import com.sleepycat.db.ReplicationManagerAckPolicy;
import com.sleepycat.db.ReplicationManagerSiteConfig;
import com.sleepycat.db.ReplicationManagerStartPolicy;
import com.sleepycat.db.ReplicationTimeoutType;
import com.sleepycat.db.VerboseConfig;

/**
 * Get a connection hopelessly clogged, and then kill the connection.
 * Verify that the blocked thread is immediately freed.
 */
public class TestDrainAbandon {
    private static final String TEST_DIR_NAME = "TESTDIR";

    private File testdir;
    private byte[] data;
    private int masterPort;
    private int clientPort;
    private int client2Port;
    private int client3Port;
    private int mgrPort;

    @Before public void setUp() throws Exception {
        testdir = new File(TEST_DIR_NAME);
        Util.rm_rf(testdir);
        testdir.mkdir();

        String alphabet = "abcdefghijklmnopqrstuvwxyz";
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        OutputStreamWriter w = new OutputStreamWriter(baos);
        while (baos.size() < 1000) // arbitrary min. size
            w.write(alphabet);
        w.close();
        data = baos.toByteArray();

        if (Boolean.getBoolean("MANUAL_FIDDLER_START")) {
            masterPort = 6000;
            clientPort = 6001;
            client2Port = 6002;
            client3Port = 6003;
            mgrPort = 8000;
        } else {
            String mgrPortNum = System.getenv("DB_TEST_FAKE_PORT");
            assertNotNull("required DB_TEST_FAKE_PORT environment variable not found",
                          mgrPortNum);
            mgrPort = Integer.parseInt(mgrPortNum);
            PortsConfig p = new PortsConfig(4);
            masterPort = p.getRealPort(0);
            clientPort = p.getRealPort(1);
            client2Port = p.getRealPort(2);
            client3Port = p.getRealPort(3);
            Util.startFiddler(p, getClass().getName(), mgrPort);
        }
    }
	
    @Test public void testDraining() throws Exception {
        EnvironmentConfig masterConfig = makeBasicConfig();
        masterConfig.setReplicationLimit(100000000);
        ReplicationManagerSiteConfig site =
            new ReplicationManagerSiteConfig("localhost", masterPort);
        site.setLocalSite(true);
        site.setLegacy(true);
        masterConfig.addReplicationManagerSite(site);

        site = new ReplicationManagerSiteConfig("localhost", clientPort);
        site.setLegacy(true);
        masterConfig.addReplicationManagerSite(site);
        site = new ReplicationManagerSiteConfig("localhost", client2Port);
        site.setLegacy(true);
        masterConfig.addReplicationManagerSite(site);
        site = new ReplicationManagerSiteConfig("localhost", client3Port);
        site.setLegacy(true);
        masterConfig.addReplicationManagerSite(site);

        Environment master = new Environment(mkdir("master"), masterConfig);
        setTimeouts(master);
        // Prevent connection retries, so that all connections
        // originate from clients 
        master.setReplicationTimeout(ReplicationTimeoutType.CONNECTION_RETRY,
                                     Integer.MAX_VALUE);
        master.replicationManagerStart(2, ReplicationManagerStartPolicy.REP_MASTER);
		
        DatabaseConfig dc = new DatabaseConfig();
        dc.setTransactional(true);
        dc.setAllowCreate(true);
        dc.setType(DatabaseType.BTREE);
        dc.setPageSize(4096);
        Database db = master.openDatabase(null, "test.db", null, dc);
		
        DatabaseEntry key = new DatabaseEntry();
        DatabaseEntry value = new DatabaseEntry();
        value.setData(data);

        for (int i=0;
             ((BtreeStats)db.getStats(null, null)).getPageCount() < 500;
             i++)
        {
            String k = "The record number is: " + i;
            key.setData(k.getBytes());
            db.put(null, key, value);
        }

        // tell fiddler to stop reading once it sees a PAGE message
        Socket s = new Socket("localhost", mgrPort);
        OutputStreamWriter w = new OutputStreamWriter(s.getOutputStream());

        String path1 = "{" + masterPort + "," + clientPort + "}"; // looks like {6000,6001}
        w.write("{init," + path1 + ",page_clog}\r\n");
        w.flush();
        BufferedReader br = new BufferedReader(new InputStreamReader(s.getInputStream()));
        br.readLine();
        assertEquals("ok", br.readLine());
        // create client
        // 
        EnvironmentConfig ec = makeBasicConfig();
        site = new ReplicationManagerSiteConfig("localhost", clientPort);
        site.setLocalSite(true);
        site.setLegacy(true);
        ec.addReplicationManagerSite(site);
        site = new ReplicationManagerSiteConfig("localhost", masterPort);
        site.setLegacy(true);
        ec.addReplicationManagerSite(site);
        site = new ReplicationManagerSiteConfig("localhost", client2Port);
        site.setLegacy(true);
        ec.addReplicationManagerSite(site);
        site = new ReplicationManagerSiteConfig("localhost", client3Port);
        site.setLegacy(true);
        ec.addReplicationManagerSite(site);
        Environment client = new Environment(mkdir("client"), ec);
        setTimeouts(client);
        client.replicationManagerStart(1, ReplicationManagerStartPolicy.REP_CLIENT);

        // wait til it gets stuck
        Thread.sleep(5000);     // FIXME

        // Do the same for another client, because the master has 2
        // msg processing threads.  (It's no longer possible to
        // configure just 1.)
        String path2 = "{" + masterPort + "," + client2Port + "}";
        w.write("{init," + path2 + ",page_clog}\r\n");
        w.flush();
        br = new BufferedReader(new InputStreamReader(s.getInputStream()));
        br.readLine();
        assertEquals("ok", br.readLine());

        ec = makeBasicConfig();
        site = new ReplicationManagerSiteConfig("localhost", client2Port);
        site.setLocalSite(true);
        site.setLegacy(true);
        ec.addReplicationManagerSite(site);
        site = new ReplicationManagerSiteConfig("localhost", masterPort);
        site.setLegacy(true);
        ec.addReplicationManagerSite(site);
        site = new ReplicationManagerSiteConfig("localhost", clientPort);
        site.setLegacy(true);
        ec.addReplicationManagerSite(site);
        site = new ReplicationManagerSiteConfig("localhost", client3Port);
        site.setLegacy(true);
        ec.addReplicationManagerSite(site);
        Environment client2 = new Environment(mkdir("client2"), ec);
        setTimeouts(client2);
        client2.replicationManagerStart(1, ReplicationManagerStartPolicy.REP_CLIENT);

        // wait til it gets stuck
        Thread.sleep(5000);


        // With the connection stuck, the master cannot write out log
        // records for new "live" transactions.  Knowing we didn't
        // write the record, we should not bother waiting for an ack
        // that cannot possibly arrive; so we should simply return
        // quickly.  The duration should be very quick, but anything
        // less than the ack timeout indicates correct behavior (in
        // case this test runs on a slow, overloaded system).
        // 
        long startTime = System.currentTimeMillis();
        key.setData("one extra record".getBytes());
        db.put(null, key, value);
        long duration = System.currentTimeMillis() - startTime;
        assertTrue("txn duration: " + duration, duration < 29000);
        System.out.println("txn duration: " + duration);
        db.close();

        // Tell fiddler to close the connections.  That should trigger
        // us to abandon the timeout.  Then create another client and
        // see that it can complete its internal init quickly.  Since
        // we have limited threads at the master, this demonstrates
        // that they were abandoned.
        //
        path1 = "{" + clientPort + "," + masterPort + "}"; // looks like {6001,6000}
        w.write("{" + path1 + ",shutdown}\r\n");
        w.flush();
        assertEquals("ok", br.readLine());
        path2 = "{" + client2Port + "," + masterPort + "}"; // looks like {6001,6000}
        w.write("{" + path2 + ",shutdown}\r\n");
        w.flush();
        assertEquals("ok", br.readLine());

        ec = makeBasicConfig();
        site = new ReplicationManagerSiteConfig("localhost", client3Port);
        site.setLocalSite(true);
        site.setLegacy(true);
        ec.addReplicationManagerSite(site);
        site = new ReplicationManagerSiteConfig("localhost", masterPort);
        site.setLegacy(true);
        ec.addReplicationManagerSite(site);
        site = new ReplicationManagerSiteConfig("localhost", clientPort);
        site.setLegacy(true);
        ec.addReplicationManagerSite(site);
        site = new ReplicationManagerSiteConfig("localhost", client2Port);
        site.setLegacy(true);
        ec.addReplicationManagerSite(site);
        
        EventHandler clientMonitor = new EventHandler();
        ec.setEventHandler(clientMonitor);
        Environment client3 = new Environment(mkdir("client3"), ec);
        setTimeouts(client3);
        startTime = System.currentTimeMillis();
        client3.replicationManagerStart(2, ReplicationManagerStartPolicy.REP_CLIENT);
        clientMonitor.await();
        duration = System.currentTimeMillis() - startTime;
        assertTrue("sync duration: " + duration, duration < 20000); // 20 seconds should be plenty

        client3.close();
        master.close();

        w.write("shutdown\r\n");
        w.flush();
        assertEquals("ok", br.readLine());
        s.close();
    }
	
    public static EnvironmentConfig makeBasicConfig() {
        EnvironmentConfig ec = new EnvironmentConfig();
        ec.setAllowCreate(true);
        ec.setInitializeCache(true);
        ec.setInitializeLocking(true);
        ec.setInitializeLogging(true);
        ec.setInitializeReplication(true);
        ec.setTransactional(true);
        ec.setThreaded(true);
        ec.setReplicationInMemory(true);
        ec.setCacheSize(256 * 1024 * 1024);
        if (Boolean.getBoolean("VERB_REPLICATION"))
            ec.setVerbose(VerboseConfig.REPLICATION, true);
        return (ec);
    }
	
    private void setTimeouts(Environment e) throws Exception {
        e.setReplicationTimeout(ReplicationTimeoutType.ACK_TIMEOUT,
                                     30000000);
    }

    public File mkdir(String dname) {
        File f = new File(testdir, dname);
        f.mkdir();
        return f;
    }
}