orbmonitor.java

CORBA orb implement

// Developed by Kinva Network Inc. 2000


// Source File Name:   OrbMonitor.java

package com.kinva.util.corb;

import com.kinva.util.misc.Log;
import java.io.*;
import java.util.*;

// Referenced classes of package com.kinva.util.corb:
//            CORBException, CORBSkeleton, Connection, Naming, 
//            Orb, ReplyQ, Request, RequestQ, 
//            Result, ServerObjectInfo

public class OrbMonitor extends Thread
{

    public OrbMonitor(int i, Connection connection, RequestQ requestq)
        throws IOException
    {
        super("OrbMonitor - " + connection);
        list = new Hashtable();
        requestQ = requestq;
        conn = connection;
        portNo = i;
        ooSt = connection.getOutputStream();
        oiSt = connection.getInputStream();
    }

    public void run()
    {
        Object obj = new Object();
        do
        {
            Object obj1;
            try
            {
                obj1 = oiSt.readObject();
                log.println(4, "OrbMonitor: got {0}: {1}", obj1, obj1.getClass());
            }
            catch(IOException ioexception)
            {
                ioexception.printStackTrace();
                if(ioexception instanceof OptionalDataException)
                {
                    OptionalDataException optionaldataexception = (OptionalDataException)ioexception;
                    log.println(1, "EOF {0} Length {1}", new Boolean(optionaldataexception.eof), new Integer(optionaldataexception.length));
                    try
                    {
                        for(int j = 0; j < optionaldataexception.length; j++)
                            log.println(1, Integer.toHexString(oiSt.read()));

                    }
                    catch(IOException ioexception1)
                    {
                        ioexception1.printStackTrace();
                    }
                }
                Date date = new Date();
                log.println(1, "{0,date} {0,time}", date);
                log.println(1, ":run(),{0} for {1}", ioexception, conn);
                Naming.removeConnection(conn.getHostAddress() + ":" + portNo);
                wakeAllReplyQs(ioexception.getMessage());
                return;
            }
            catch(ClassNotFoundException classnotfoundexception)
            {
                obj1 = null;
                classnotfoundexception.printStackTrace();
                log.println(1, ":run(),{0}", classnotfoundexception.getMessage());
            }
            if(obj1 instanceof Request)
            {
                Request request = (Request)obj1;
                ServerObjectId serverobjectid = request.getServerId();
                log.println(4, "Getting Server Object Info for : {0}", serverobjectid);
                ServerObjectInfo serverobjectinfo = Naming.getServerObjectInfo(serverobjectid);
                log.println(4, "Got Server Object Info for : {0}", serverobjectid);
                Object obj2 = null;
                if(serverobjectinfo == null)
                {
                    log.println(1, "No Server Object Info for : {0}", serverobjectid);
                    if(!request.isOneWay())
                    {
                        Result result1 = new Result(request.getRequestId());
                        result1.setStatus(2);
                        result1.setResult(new CORBException("Server object not bound"));
                        writeResult(result1);
                    }
                    return;
                }
                CORBSkeleton corbskeleton = null;
                RequestQ requestq = null;
                try
                {
                    corbskeleton = (CORBSkeleton)serverobjectinfo.getSkeleton();
                    requestq = corbskeleton.getRequestQ();
                }
                catch(CORBException corbexception)
                {
                    log.println(4, "Could not get skeleton for : {0}", serverobjectid);
                    if(!request.isOneWay())
                    {
                        Result result2 = new Result(request.getRequestId());
                        result2.setStatus(2);
                        result2.setResult(corbexception);
                        writeResult(result2);
                    }
                    return;
                }
                log.println(3, "Resetting heartbeat : {0}:{1}", serverobjectinfo, corbskeleton);
                serverobjectinfo.setLastHeartBeat(System.currentTimeMillis());
                serverobjectinfo.setTimeout(0);
                if(requestq != null)
                {
                    request.setOutputStream(ooSt);
                    requestq.enqueue(request);
                } else
                {
                    request.setOutputStream(ooSt);
                    requestQ.enqueue(request);
                    log.println(5, "Request {0} enqueued", request);
                }
            } else
            if(obj1 instanceof Result)
            {
                Result result = (Result)obj1;
                int i = result.getRequestId();
                ReplyQ replyq = (ReplyQ)list.get(new Integer(i));
                if(replyq != null)
                    replyq.enqueue(result);
                else
                    log.println(1, "No ReplyQ for " + i);
            }
        } while(true);
    }

    public void registerQueue(int i, ReplyQ replyq)
    {
        list.put(new Integer(i), replyq);
    }

    public void unregisterQueue(int i)
    {
        list.remove(new Integer(i));
    }

    private void writeResult(Result result)
    {
        try
        {
            synchronized(ooSt)
            {
                ooSt.writeObject(result);
                ooSt.flush();
                ooSt.reset();
            }
            return;
        }
        catch(IOException ioexception)
        {
            log.println(1, "{0}: Unable to write to output stream for {1}", Thread.currentThread(), conn);
            ioexception.printStackTrace();
            return;
        }
    }

    private void wakeAllReplyQs(String s)
    {
        Hashtable hashtable = (Hashtable)list.clone();
        ReplyQ replyq;
        Result result;
        for(Enumeration enumeration = hashtable.keys(); enumeration.hasMoreElements(); replyq.enqueue(result))
        {
            Integer integer = (Integer)enumeration.nextElement();
            replyq = (ReplyQ)hashtable.get(integer);
            result = new Result(integer.intValue());
            result.setStatus(2);
            if(s == null)
                result.setResult(new CORBException("Connection lost"));
            else
                result.setResult(new CORBException(s));
            result.flush();
        }

    }

    private Hashtable list;
    static Log log;
    private ObjectInputStream oiSt;
    private ObjectOutputStream ooSt;
    private RequestQ requestQ;
    private Connection conn;
    private int portNo;

    static 
    {
        log = Orb.log;
    }
}