simplexmlserver.java

High performance DB query

/*
 * @(#)$Id: SimpleXMLServer.java,v 1.5 2004/07/02 23:59:21 huebsch Exp $
 *
 * Copyright (c) 2001-2004 Regents of the University of California.
 * All rights reserved.
 *
 * This file is distributed under the terms in the attached BERKELEY-LICENSE
 * file. If you do not find these files, copies can be found by writing to:
 * Computer Science Division, Database Group, Universite of California,
 * 617 Soda Hall #1776, Berkeley, CA 94720-1776. Attention: Berkeley License
 *
 * Copyright (c) 2003-2004 Intel Corporation. All rights reserved.
 *
 * This file is distributed under the terms in the attached INTEL-LICENSE file.
 * If you do not find these files, copies can be found by writing to:
 * Intel Research Berkeley, 2150 Shattuck Avenue, Suite 1300,
 * Berkeley, CA, 94704.  Attention:  Intel License Inquiry.
 */


package pier.helpers.handlers;

import java.io.ByteArrayInputStream;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import org.apache.log4j.Logger;
import pier.components.PierFrontend;
import pier.data.Tuple;
import pier.data.TupleCollection;
import pier.helpers.formatters.Formatter;
import pier.helpers.formatters.XMLFormatter;
import pier.query.QueryPlan;
import pier.query.QueryTag;
import services.LocalNode;

/**
 * Class SimpleXMLServer
 */
public class SimpleXMLServer extends SimpleServer {

    private static Logger logger = Logger.getLogger(SimpleXMLServer.class);

    /**
     * Constructor SimpleXMLServer
     */
    public SimpleXMLServer() {
        super();

        this.formatter = new XMLFormatter();
    }

    /**
     * Method processOptions
     */
    public void processOptions() {

        // Replace (non slash), with \n
        options = options.replaceAll("[^\\\\]\\u002C", "\n");

        // Replace \, with ,
        options = options.replaceAll("\\\\,", ",");

        // Replace \h with #
        options = options.replaceAll("\\\\h", "#");

        try {
            optionsTable.load(new ByteArrayInputStream(options.getBytes()));
        } catch (Exception exception) {
            error(PierFrontend.createErrorFromThrowable(exception), refnum,
                  null, null);
        }
    }

    /**
     * Method startQuery
     */
    public void startQuery() {
        queryStartTime = (long) (LocalNode.myTimer.getCurrentTime() * 1000);

        double timeoutValue =
            Double.parseDouble(optionsTable.getProperty(TIMEOUT_KEY,
                                                        DEFAULT_TIMEOUT));

        if (timeoutValue > 0) {
            LocalNode.myTimer.schedule(timeoutValue, SIGNAL_TIMEOUT, this);
        }

        maxTuples = Integer.parseInt(optionsTable.getProperty(MAXTUPLES_KEY,
                DEFAULT_MAXTUPLES));
        maxBytes = Integer.parseInt(optionsTable.getProperty(MAXBYTES_KEY,
                DEFAULT_MAXBYTES));
        timestampType = Integer.parseInt(optionsTable.getProperty(TIMESTAMP_KEY,
                DEFAULT_TIMESTAMP));

        QueryPlan thePlan = parent.startQuery(query, refnum, this);

        connection.write(
            formatter.messagePreamble(
                Formatter.MESSAGE_TYPE_RESULT, currentTimestamp()));
    }

    /**
     * Method result
     *
     * @param tuples
     * @param refnum
     * @param queryTag
     * @param source
     */
    public void result(TupleCollection tuples, int refnum, QueryTag queryTag,
                       InetSocketAddress source) {
        if ((curState == STATE_QUERY_STARTED)
                && (curState != STATE_CONNECTION_CLOSED)) {
            int numTuples = tuples.size();

            for (int i = 0; i < numTuples; i++) {
                Tuple theTuple = tuples.getTuple(i);
                ByteBuffer tupleBuffer =
                    formatter.formatTuple(theTuple, refnum, queryTag, source,
                                          currentTimestamp());
                int tupleBytes = tupleBuffer.limit();

                connection.write(tupleBuffer);

                if (maxTuples > 0) {
                    maxTuples--;

                    if (maxTuples == 0) {
                        curState = STATE_QUERY_ENDED;

                        connection.write(
                            formatter.messagePostamble(
                                Formatter.MESSAGE_TYPE_RESULT,
                                currentTimestamp()));
                        closeConnection();

                        break;
                    }
                }

                if (maxBytes > 0) {
                    maxBytes -= tupleBytes;

                    if (maxBytes <= 0) {
                        curState = STATE_QUERY_ENDED;

                        connection.write(
                            formatter.messagePostamble(
                                Formatter.MESSAGE_TYPE_RESULT,
                                currentTimestamp()));
                        closeConnection();

                        break;
                    }
                }
            }
        }
    }

    /**
     * Method handleClock
     *
     * @param clockData
     */
    public void handleClock(Object clockData) {
        if ((clockData.equals(SIGNAL_TIMEOUT))
                && (curState != STATE_CONNECTION_CLOSED)) {
            curState = STATE_QUERY_ENDED;

            connection.write(
                formatter.messagePostamble(
                    Formatter.MESSAGE_TYPE_RESULT, currentTimestamp()));
            closeConnection();
        }
    }
}