agent.java

cqME :java framework for TCK test.

/*
 * $Id$
 *
 * Copyright 1996-2007 Sun Microsystems, Inc. All Rights Reserved.
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License version
 * 2 only, as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
 * General Public License version 2 for more details (a copy is
 * included at /legal/license.txt).
 *
 * You should have received a copy of the GNU General Public License
 * version 2 along with this work; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
 * 02110-1301 USA
 *
 * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa
 * Clara, CA 95054 or visit www.sun.com if you need additional
 * information or have any questions.
 *
 */

package com.sun.tck.j2me.execution.client;

import java.io.DataOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Vector;

import com.sun.javatest.Status;
import com.sun.tck.j2me.execution.baseclient.BaseAgent;
import com.sun.tck.j2me.execution.baseclient.BaseExecProtocol;
import com.sun.tck.j2me.execution.baseclient.Constants;
import com.sun.tck.j2me.execution.baseclient.ExecutionRequest;
import com.sun.tck.j2me.execution.client.ExecutionProtocol.ProactiveCommandProcessor;

/**
 * The agent is responsible for the execution of the request from the
 * JavaTest and delivering of the results to the JavaTest.
 * <p>
 * Note that the agent is not responsible for the communication
 * between JavaTest and the Agent. It uses
 * {@link com.sun.tck.j2me.services.commService.CommToJavatestService
 * CommToJavatestService}, which should be configured prior the Agent
 * usage.
 * <p>
 * The agent functionality is split into two parts. The base
 * functionality is implemented by the {@link BaseAgent}. It includes
 * only execution of the tests.<br>
 * The Agent additionally supports dynamic agent registration,
 * {@link com.sun.tck.j2me.execution.server.ServerProactiveCommand
 * proactive commands}.
 */
public class Agent extends BaseAgent implements TraceLog, Runnable {
    private static final String PPROACTIVE_COMMAND_PROCESSED =
            "com.sun.j2me.execution.proactive_commands.processed";

    private static final long TIMEOUT = 10 * 1000L;

    /**
     * The registrator is responsible for maintaining of the
     * registration of the agents.
     */
    protected Registrator registrator = new Registrator(15);

    private int concurrency = 1;

    private String map;

    private Executor[] executors;

    private RegistrationTicket ticket;

    private Object ticketMonitor = new Object();

    private Hashtable properties;

    private volatile boolean isRunning;

    /**
     * This class is responsible for the execution of the requests in
     * separate thread.
     */
    protected class Executor extends Thread {
        private volatile boolean isRunning;
        private String id;

        /**
         * Creates a Executor with the given id
         */
        public Executor(String id) {
            this.id = id;
        }

        public void run() {
            isRunning = true;
            try {
                executeRequests(id);
            } finally {
                isRunning = false;
                completeExecutor();
            }
        }
        
        public boolean isRunning() {
            return isRunning;
        }
    }

    /**
     * Returns agent ID. If the current registration is not valid,
     * this method initiates communication with JavaTest to update or
     * get a valid registration.
     */
    protected String getAgentId() {
        return checkRegistration(false).getId();
    }

    /**
     * Creates a Agent with the given command and concurrency.
     */
    public Agent(PlatformAdapter command, int concurrency) {
        this(command, concurrency, "", "ServerExecutionService");
    }

    /**
     * Creates a Agent. The CommToJavatestService service should be
     * initialized prior agent creation.
     */
    public Agent(PlatformAdapter command, int concurrency, String agentId,
            String serviceName) {
        super(agentId, command, serviceName);
        if (concurrency < 1) {
            throw new IllegalArgumentException(
                    "The concurrency mist be greater than 0");
        }
        this.concurrency = concurrency;
        super.setObserver(new ListOfObservers());
    }

    /**
     * Sets the translation map to be used to localize incoming
     * requests. This map will be sent to the JavaTest during
     * registration.
     *
     * @param in
     *                input stream containing map.
     * @param encoding
     *                used for encoding of the bytes to the characters
     *                or null of the default encoding should be used.
     */
    public synchronized void setMap(InputStream in, String encoding)
            throws IOException {
        InputStreamReader reader =
                ((encoding == null) ? new InputStreamReader(in)
                        : new InputStreamReader(in, encoding));
        StringBuffer mapBuff = new StringBuffer();
        char[] buff = new char[8 * 1024];
        int read = 0;
        while ((read = reader.read(buff)) >= 0) {
            mapBuff.append(new String(buff, 0, read));
        }
        map = mapBuff.toString();
    }

    /**
     * Returns the Notifier. This method never return null.
     */
    public ListOfObservers getObservers() {
        if (!(getObserver() instanceof ListOfObservers)) {
            super.setObserver(new ListOfObservers());
        }
        return (ListOfObservers) super.getObserver();
    }

    /**
     * Sets communication protocol.
     *
     * @throws IllegalArgumentException
     *                 if the given protocol is not instance of
     *                 ExecutionProtocol.
     */
    public void setExecutionProtocol(BaseExecProtocol protocol)
            throws IllegalArgumentException {
        if (!(protocol instanceof ExecutionProtocol)) {
            throw new IllegalArgumentException("ExecutionProtocol should be "
                    + "instance of ExecutionProtocol");
        }
        ((ExecutionProtocol) protocol)
                .setProactiveCommandProcessor(new ProactiveCommandProcessor() {
                    public void process(ExecutionRequest cmd,
                            DataOutputStream out) {
                        if (Constants.REREGISTER.equals(cmd.getName())) {
                            String processed =
                                    (String) properties.get(
                                            PPROACTIVE_COMMAND_PROCESSED);
                            properties.put(PPROACTIVE_COMMAND_PROCESSED,
                                    (processed == null) ? cmd.getId()
                                            : (processed + " " + cmd.getId()));
                            checkRegistration(true);
                        }
                    }

                    public boolean doesSupport(ExecutionRequest cmd) {
                        return Constants.REREGISTER.equals(cmd.getName());
                    }
                });
        super.setExecutionProtocol(protocol);
    }

    /**
     * Runs the agent. Since an Agent is {@link Runnable runnable},
     * this method will typically be called on a separate thread.
     */
    public synchronized void run() {
        if (getExecutionService() == null) {
            setExecutionProtocol(new ClientExecutionService());
        }
        if (isRunning) {
            throw new RuntimeException("Agent already running");
        }
        isRunning = true;
        start();
        if (getObserver() != null) {
            getObserver().started();
        }
        registrator.setExecutionProtocol((ExecutionProtocol) getExecutionService());
        registrator.start();
        try {
            trace("AGENT STARTED, maxThreads=" + concurrency);
            checkRegistration(false);
            executors = new Executor[concurrency];
            for (int i = 0; i < concurrency; i++) {
                String executorName = getAgentId() + ":" + nextAgentNumber();
                executors[i] = new Executor(executorName);
                executors[i].start();
                trace("EXECUTOR " + executors[i].id + " STARTED "
                        + getClass().getName());
            }
            waitUntilDone();
        } catch (InterruptedException e) {
            return;
        } finally {
            if (getObserver() != null) {
                getObserver().finished();
            }
            trace("AGENT EXITING");
            registrator.stop();
            stop();
            isRunning = false;
        }
    }

    /**
     * Checks that the registration is valid and if it is not or force
     * is true the method requests re-registration from the JavaTest.
     *
     * @param force
     *                true means that the re-registration should be
     *                requested even if the current registration is
     *                valid.
     */
    protected RegistrationTicket checkRegistration(boolean force) {
        String proposedID = "";
        synchronized (ticketMonitor) {
            if ((ticket != null) && (ticket.isValid()) && !force) {
                return ticket;
            }
            if (ticket != null) {
                proposedID = ticket.getId();
                registrator.remove(ticket.getId());
            }

            ticket = null;
            Hashtable props = createProperties();
            while (ticket == null) {
                try {
                    ticket = ((ExecutionProtocol) getExecutionService())
                            .register(proposedID, props);
                    registrator.addAgent(ticket);
                } catch (IOException e) {
                    // TODO
                }
            }
            trace("Agent has been registered:" + ticket.getId());
            return ticket;
        }
    }

    /**
     * Creates the registration properties. These properties are send
     * to the JavaTest side during initial registration.
     */
    protected synchronized Hashtable createProperties() {
        if (properties == null) {
            properties = new Hashtable();
        }
        if (map != null) {
            properties.put(Constants.AGENT_MAP, map);
        }
        return properties;
    }

    /**
     * Sets the registration properties.
     */
    public synchronized void setRegistartionProperties(Hashtable props) {
        properties = props;
    }

    private void completeExecutor() {
        synchronized (this) {
            for (int i = 0; i < executors.length; i++) {
                if (executors[i].isRunning()) {
                    return;
                }
            }
            isRunning = false;
            this.notifyAll();
        }
    }

    /**
     * Adds registration observer. The observer will receive notification about
     * registration related events.
     */
    public void addRegistrationObserver(RegistrationObserver observer) {
        registrator.addObserver(observer);
    }

    private void waitUntilDone() throws InterruptedException {
        while (isRunning) {
            synchronized (this) {
                this.wait(TIMEOUT);
            }
        }
    }

    /**
     * Implementation of the Observer, which allows to register others
     * Observers. The methods are dispatched to the all registered
     * Observers implementation in order of registration.
     */
    public class ListOfObservers implements Observer {
        private Vector observers = new Vector();

        /**
         * Adds a new Observer to the list. If the Observer is already
         * registered, the method do nothing.
         */
        public synchronized void addObserver(Observer observer) {
            if ((observer != null) && !observers.contains(observer)) {
                observers.addElement(observer);
            }
        }

        /**
         * Removes the given Observer from the list.
         */
        public synchronized void removeObserver(Observer observer) {
            observers.removeElement(observer);
        }

        public synchronized void started() {
            for (Enumeration e = observers.elements(); e.hasMoreElements();) {
                ((Observer) e.nextElement()).started();
            }
        }

        public void finished() {
            for (Enumeration e = observers.elements(); e.hasMoreElements();) {
                ((Observer) e.nextElement()).finished();
            }
        }

        public void requestStarted(String executorId, ExecutionRequest request) {
            for (Enumeration e = observers.elements(); e.hasMoreElements();) {
                ((Observer) e.nextElement()).requestStarted(
                        executorId, request);
            }
        }

        public void requestCompleted(String executorId,
                ExecutionRequest request, Status status) {
            for (Enumeration e = observers.elements(); e.hasMoreElements();) {
                ((Observer) e.nextElement()).requestCompleted(executorId,
                        request, status);
            }
        }
    }

    /* For autonumbering agent tasks. */
    private static int agentInitNumber;

    private static synchronized int nextAgentNumber() {
        return agentInitNumber++;
    }
}