java/ldapjdk/EclReadAndPlay.java

/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License, Version 1.0 only
 * (the "License").  You may not use this file except in compliance
 * with the License.
 *
 * You can obtain a copy of the license at
 * trunk/opends/resource/legal-notices/OpenDS.LICENSE
 * or https://OpenDS.dev.java.net/OpenDS.LICENSE.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at
 * trunk/opends/resource/legal-notices/OpenDS.LICENSE.  If applicable,
 * add the following below this CDDL HEADER, with the fields enclosed
 * by brackets "[]" replaced with your own identifying information:
 *      Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 *
 *
 *      Copyright 2010 Sun Microsystems, Inc.
 */

import netscape.ldap.*;
import netscape.ldap.util.*;

import java.util.*;
import java.io.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

class EclReadAndPlay {

    // dbPath --> stores files containing csn of identifiers
    static final String dbPath = "."; /*"db";*/
    // configPath --> stores masters config file
    static final String configPath = "."; /*"config";*/
    static final String mastersFilename = "masters";
    static final String logsPath = "."; /*"logs";*/
    static final String accessFilename = "access";

    // Maximum time (in milliseconds) without update being read from the master: 60 s
    static final int MAX_IDLE_TIME = 60000;

    // ECL "draft" mode -- Initial changeNumber
    static final int INITIAL_CHANGENUMBER = 1;
    // ECL "opends" mode -- Initial control value:
    // --control "1.3.6.1.4.1.26027.1.5.4:false:;" ==> first cookie: ";"
    static final String INITIAL_COOKIE = ";";


    static PrintWriter standardOut = null;
    static PrintWriter accessOut = null;
    static HashMap<String,CSN> RUV;
    static Object lock;
    static HashMap<String,File> files;
    static int nb_ops = 0;
    static int total_nb_ops = 0;
    static int nb_ignored = 0;
    static int changeNumber = 0;
    static int lastChangeNumber = 0;
    static String changelogCookie = null;
    static String lastExternalChangelogCookie = null;
    static int missingChanges = 0;
    static int lastMissingChanges = 1;

    static String eclMode = null;
    static int queueSize = 0;
    static String bindDn = null;
    static String bindPwd = null;
    static boolean displayMissingChanges = false;
    static String outputFilename = null;

    public static void main( String[] args )

   {


        FileWriter out = null;

        files = new HashMap<String,File>();

        // Load latest read CSN values from files in db/ directory
        File csnDir = new File(dbPath);
        RUV = new HashMap<String,CSN>();
        try {
            FilenameFilter csnFileFilter = new FilenameFilter() {
                public boolean accept(File dir, String name) {
                    return name.endsWith(".csn");
                }
            };
            File[] csnFiles = csnDir.listFiles(csnFileFilter);
            if ( csnFiles != null ) {
                for (File f: csnFiles) {
                    String csnFilename = f.getName();
                    String id =
                        csnFilename.substring(0, csnFilename.indexOf(".csn"));
                    BufferedReader in = new BufferedReader (new FileReader(f));
                    CSN mycsn = new CSN(in.readLine());
                    if ( mycsn.value == null )
                        mycsn = new CSN("00000000000000000000");
                    // System.out.println(files[i] + "\t" + mycsn);
                    RUV.put(id, mycsn);

                    files.put(id, f);
                }
            }
        } catch (IOException e) {
            println("ERROR", e.toString());
            e.printStackTrace();
            System.exit(1);
        }


        /********** Parse arguments **********/
        int masterN=0;
        String hostport = null;
        ArrayList<Server> masters = new ArrayList<Server>();

        for (int k = 0; k < args.length; k++) {
            String opt = args[k];
            String val = args[k+1];

            // ECL mode: "opends" or "draft"
            if (opt.equals("-m")) {
                eclMode = val;
            }

            // Queue size
            if (opt.equals("-q")) {
                queueSize = Integer.parseInt(val);
            }

            // Display missing changes?
            if (opt.equals("-x")) {
                if ( val.equals("true") )
                    displayMissingChanges = true;
                else
                    displayMissingChanges = false;
            }

            // Bind DN
            if (opt.equals("-D")) {
                bindDn = val;
                System.out.println(".......... bindDN: " + bindDn);
            }

            // Bind password
            if (opt.equals("-w")) {
                bindPwd = val;
                System.out.println(".......... bindPwd: " + bindPwd);
            }


            // Stand-alone server:port
            if (opt.equals("-s")) {
                hostport = val;
                System.out.println(".......... stand-alone server: " + hostport);
            }


            // Replicated masters
            if (opt.equals("-p")) {
                masters.add(new Server(val));
            }


            // Standard output file
            if (opt.equals("-o")) {
                outputFilename = val;
            }

            k++;
        } /* for() */


        if ( eclMode == null || queueSize == 0 || bindDn == null ||
             bindPwd == null || hostport == null || masters.size() == 0 ||
             outputFilename == null ) {
            System.out.println("usage: -m {draft|opends} -q {queue size} "
                               + "-x {(displayMissingChanges):true|false} "
                               + "-o {outputFilename} -D {bindD} -w {bindPwd} "
                               + "-s {standalone-host:port} "
                               + "-p {master1-host:port} "
                               + "-p {master2-host:port}...");
            System.exit(1);
        }

        /* try {
            File mastersFile= new File(configPath, mastersFilename);
            LineNumberReader in=new LineNumberReader (new FileReader(mastersFile) );
            String line;
            while ( in.ready() ) {
                line=in.readLine();
                masters.add(new Server(line));
            }
        } catch (IOException e) {
            println ("ERROR", e.toString());
            System.exit(1);
        } */

        masters.trimToSize();
        // System.out.println(masters);

        /*************************************/


        // Output file (logs are appended)
        try {
            standardOut = new PrintWriter(new BufferedWriter(new FileWriter( new File(outputFilename)) ) );
        } catch (IOException e) {
            println ("ERROR", e.toString() );
            e.printStackTrace();
            System.exit(1);
        }


        // Access log (data is appended)
        try {
            accessOut = new PrintWriter(new BufferedWriter(new FileWriter( new File(logsPath, accessFilename)) ) );
        } catch (IOException e) {
            println ("ERROR", e.toString() );
            e.printStackTrace();
            System.exit(1);
        }


        /********** Initialise reader/writer threads **********/
        // Create a bounded blocking queue of integers
        BlockingQueue<Change> queue = new ArrayBlockingQueue<Change>(queueSize);

        // Initialise reader thread --> read updates from replicated master
        Reader reader = new Reader(queue, masters);
        reader.start();

        // Initialise writer thread --> write updates onto stand-alone server
        Writer writer = new Writer(queue, hostport);
        writer.start();

        lock = new Object();
        synchronized (lock) {
            try {
              lock.wait();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }


        long start = System.currentTimeMillis();
        int i=0;
        while (true) {
            int loopPeriod = 10; /* 10 s */
            sleep(loopPeriod * 1000);
            total_nb_ops += nb_ops;
            long duration = ((System.currentTimeMillis() - start)/1000);
            println("INFO", "Replayed " + nb_ops/loopPeriod
                    + " ops/sec. (Avg = " + (total_nb_ops/duration)
                    + " ops/s, Total = " + total_nb_ops
                    + " , ignored = " + nb_ignored + " )");
            nb_ops = 0;
            if ( i++ == 3 && displayMissingChanges == true ) {
                if ( eclMode.equals("draft") ) {
                    missingChanges = lastChangeNumber - changeNumber;
                    float percentage = (lastMissingChanges - missingChanges);
                    println("INFO", "Current changeNumber = " + changeNumber
                            + ", lastChangeNumber = " + lastChangeNumber
                            + ", missing changes = " + missingChanges + "/"
                            + lastMissingChanges + " (" + percentage + ")");
                    lastMissingChanges = missingChanges;
                    if (lastMissingChanges == 0)
                        lastMissingChanges = 1;
                } else if ( eclMode.equals("opends") ) {
                    println("INFO", "Current changelogCookie = "
                            + changelogCookie
                            + ", lastExternalChangelogCookie = "
                            + lastExternalChangelogCookie);
                }
                i = 0;
            }
        }

    }

     public static void inc_ops(int c) {
        nb_ops++;
        changeNumber = c;
    }

     public static void inc_ops(String c) {
         nb_ops++;
         changelogCookie = c;
     }

    public static void inc_ignored(int c) {
        nb_ignored++;
        changeNumber = c;
    }

    public static void inc_ignored(String c) {
        nb_ignored++;
        changelogCookie = c;
    }

    public static String getDate() {

        // Initialize the today's date string
        String DATE_FORMAT = "yyyy/MM/dd:HH:mm:ss";
        java.text.SimpleDateFormat sdf =
            new java.text.SimpleDateFormat(DATE_FORMAT);
        Calendar c1 = Calendar.getInstance(); // today
        return("[" + sdf.format(c1.getTime()) + "]");
   }

   public static void println(String level, String msg) {
        standardOut.println (getDate() + " - " + level + ": " + msg );
   }

   public static void sleep(int time) {
        try {
            Thread.sleep(time);
        }
        catch ( InterruptedException e )
        {
             println( "ERROR" ,  e.toString() );
        }
    }
}