reentrantthreadhandlerimpl.java

来自「SRI international 发布的OAA框架软件」· Java 代码 · 共 102 行

/* $Id: ReentrantThreadHandlerImpl.java,v 1.2 2005/03/29 02:13:24 giuli Exp $*/
/**************************************************************************
 * Copyright 2002, 2003 SRI International. All rights reserved.
 *
 * The material contained in this file is confidential and proprietary to SRI
 * International and may not be reproduced, published, or disclosed to others
 * without authorization from SRI International.
 *
 * DISCLAIMER OF WARRANTIES
 *
 * SRI International MAKES NO REPRESENTATIONS OR WARRANTIES ABOUT THE
 * SUITABILITY OF THE SOFTWARE, EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT
 * LIMITED TO THE IMPLIED WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
 * PARTICULAR PURPOSE, OR NON-INFRINGEMENT. SRI International SHALL NOT BE
 * LIABLE FOR ANY DAMAGES SUFFERED BY LICENSEE AS A RESULT OF USING, MODIFYING
 * OR DISTRIBUTING THIS SOFTWARE
 **************************************************************************/
package com.sri.sedc.javanetbridge;

import com.sri.sedc.javanetbridge.io.DataMarshalOutputStream;

import java.io.IOException;

import edu.emory.mathcs.backport.java.util.concurrent.LinkedBlockingQueue;

/**
 * Default implementation of ReentrantThreadHandler.
 */
class ReentrantThreadHandlerImpl implements ReentrantThreadHandler, Runnable {
    private JavaNetBridgeBean bean;
    private LinkedBlockingQueue queue;
    private DataMarshalOutputStream out;
    private int threadId;
    private volatile boolean running = false;

    ReentrantThreadHandlerImpl(JavaNetBridgeBean bean, DataMarshalOutputStream out, int threadId) {
        this.bean = bean;
        this.out = out;
        this.threadId = threadId;
        queue = new LinkedBlockingQueue();
    }

    public void queueMethodCall(int methodId, Object[] args) throws InterruptedException {
        queue.put(new Object[] { new Integer(methodId), args});
    }

    public int getRemoteServerId() {
        return 0;
    }

    public void start() {
        running = true;
        new Thread(this).start();
    }

    public void stop() {
        running = false;
    }

    public void run() {
        boolean wasInterrupted = false;
        while (running) {
            try {
                Object[] nextMethod = (Object[])queue.take();
                int methodId = ((Integer)nextMethod[0]).intValue();
                Object[] args = (Object[])nextMethod[1];
                bean.invokeMethod(methodId, args, out, threadId);
            } catch (InterruptedException ex) {
                wasInterrupted = true;
                running = false;
                ex.printStackTrace();
            } catch (Exception ex) {
                ex.printStackTrace();
                synchronized (out) {
                    try {
                        out.write(threadId);
                        out.write(JavaNetBridgeBean.METHOD_INVOKE_EXCEPTION);
                        out.write(ex.getClass().getName());
                        out.write(ex.getMessage());
                        out.flush();
                    } catch (IOException e) {
                        e.printStackTrace();
                        running = false;
                    }
                }
            }
        }

        try {
            while (!queue.isEmpty()) {
                queue.take();
            }
        } catch (Exception ex) {
            // Ignore...
        }

        if (wasInterrupted) {
            Thread.currentThread().interrupt();
        }
    }
}