orb.java

orb implement

// Developed by Kinva Network Inc. 2000


// Source File Name:   Orb.java

package com.kinva.util.corb;

import com.kinva.cyberman.api.misc.Config;
import com.kinva.util.misc.Log;
import com.kinva.util.net.*;
import java.applet.Applet;
import java.io.*;
import java.net.*;

// Referenced classes of package com.kinva.util.corb:
//            CORBException, CORBRuntimeException, CORBSkeleton, CORBStub, 
//            Connection, GarbageCollectionInfo, GarbageCollector, HeartBeatImpl, 
//            HeartBeatThread, LookupImpl, Naming, OrbMonitor, 
//            RObject, ReplyQ, Request, RequestQ, 
//            Result, ServerObjectId, ServerObjectInfo

public class Orb
    implements Runnable
{

    Orb()
    {
    }

    public static synchronized void startup(Applet applet)
        throws IOException, DottedQuadFormatException, UnknownHostException, ClassNotFoundException, CORBException
    {
        if(started)
            return;
        log.println(1, "Starting Orb ...");
        URL url = applet.getDocumentBase();
        String s = url.getHost();
        int i = 0;
        try
        {
            i = Integer.parseInt(Config.getParam("corb.root.port"));
        }
        catch(Exception _ex)
        {
            i = Integer.parseInt("9996");
        }
        Connection connection = new Connection(s, i);
        ObjectOutputStream objectoutputstream = connection.getOutputStream();
        ObjectInputStream objectinputstream = connection.getInputStream();
        ipAddr = (IpAddress)objectinputstream.readObject();
        localPort = connection.getSocket().getLocalPort();
        new Naming(ipAddr, localPort);
        LookupImpl lookupimpl = new LookupImpl();
        ServerObjectId serverobjectid = exportObject(lookupimpl, 1);
        RObject robject = Naming.getSkeleton(serverobjectid.getServerId());
        robject.getGarbageCollectionInfo().setLiveForEver(true);
        HeartBeatImpl heartbeatimpl = new HeartBeatImpl();
        serverobjectid = exportObject(heartbeatimpl, 2);
        robject = Naming.getSkeleton(serverobjectid.getServerId());
        robject.getGarbageCollectionInfo().setLiveForEver(true);
        CORBStub corbstub = (CORBStub)((CORBSkeleton)robject).getStub();
        corbstub.setOneWay(true);
        requestQ = new RequestQ(log);
        requestQ.start();
        RObject robject1 = (RObject)objectinputstream.readObject();
        RObject robject2 = (RObject)objectinputstream.readObject();
        objectoutputstream.writeObject(new Integer(localPort));
        objectoutputstream.flush();
        objectoutputstream.reset();
        objectoutputstream.writeObject(Naming.getStub(1));
        objectoutputstream.writeObject(Naming.getStub(2));
        objectoutputstream.flush();
        OrbMonitor orbmonitor = new OrbMonitor(connection.getPortNumber(), connection, requestQ);
        connection.setOrbMonitor(orbmonitor);
        Naming.setAppletAddrMapping(s, robject1.getServerObjectId().getIpAddress());
        Naming.addStub(connection.getId(), connection, robject1);
        Naming.addStub(connection.getId(), connection, robject2);
        orbmonitor.start();
        log.println(1, "Orb started...");
        log.println(1, "Connected from " + ipAddr + ":" + localPort + " to " + s + ":" + i);
        heartBeatThread = new HeartBeatThread();
        garbageCollector = new GarbageCollector();
        started = true;
    }

    public static synchronized void startup()
        throws IOException, DottedQuadFormatException, UnknownHostException, CORBException
    {
        startup(0);
    }

    public static synchronized void startup(int i)
        throws IOException, DottedQuadFormatException, UnknownHostException, CORBException
    {
        if(started)
        {
            return;
        } else
        {
            serverSock = new ServerSocket(i);
            InetAddress inetaddress = InetAddress.getLocalHost();
            ipAddr = new IpAddress(inetaddress.getAddress());
            System.out.println("ipAddr = " + ipAddr);
            localPort = serverSock.getLocalPort();
            new Naming(ipAddr, localPort);
            LookupImpl lookupimpl = new LookupImpl();
            ServerObjectId serverobjectid = exportObject(lookupimpl, 1);
            RObject robject = Naming.getSkeleton(serverobjectid.getServerId());
            robject.getGarbageCollectionInfo().setLiveForEver(true);
            HeartBeatImpl heartbeatimpl = new HeartBeatImpl();
            serverobjectid = exportObject(heartbeatimpl, 2);
            robject = Naming.getSkeleton(serverobjectid.getServerId());
            robject.getGarbageCollectionInfo().setLiveForEver(true);
            CORBStub corbstub = (CORBStub)((CORBSkeleton)robject).getStub();
            corbstub.setOneWay(true);
            requestQ = new RequestQ(log);
            requestQ.start();
            orbThread = new Thread(new Orb());
            orbThread.setName("CORBThread");
            orbThread.start();
            log.println(1, "Orb started...");
            log.println(1, "Listening on " + ipAddr + ":" + localPort);
            heartBeatThread = new HeartBeatThread();
            garbageCollector = new GarbageCollector();
            started = true;
            return;
        }
    }

    public static synchronized void shutdown()
        throws CORBException
    {
        if(!started)
            throw new CORBException("Not started yet");
        started = false;
        heartBeatThread.stop();
        garbageCollector.stop();
        requestQ.shutdown();
        requestQ.stop();
        if(orbThread != null)
            orbThread.stop();
        ServerObjectInfo aserverobjectinfo[] = Naming.getBoundObjects();
        for(int i = 0; i < aserverobjectinfo.length; i++)
        {
            CORBSkeleton corbskeleton = null;
            aserverobjectinfo[i].getServerObjectId();
            try
            {
                corbskeleton = (CORBSkeleton)aserverobjectinfo[i].getSkeleton();
            }
            catch(CORBException corbexception)
            {
                corbexception.printStackTrace();
                continue;
            }
            RequestQ requestq = corbskeleton.getRequestQ();
            if(requestq != null)
            {
                requestq.shutdown();
                requestq.stop();
            }
        }

        Naming.shutdown();
        log.println(1, "Orb shutdown...");
    }

    public static String getAddress()
    {
        if(ipAddr != null)
            return ipAddr.toString();
        else
            return null;
    }

    public static int getPort()
    {
        if(serverSock != null)
            return serverSock.getLocalPort();
        else
            return -1;
    }

    public static ServerObjectId exportObject(Object obj)
        throws CORBException
    {
        return Naming.bind(obj);
    }

    public static ServerObjectId exportObject(Object obj, int i)
        throws CORBException
    {
        return Naming.bind(obj, i);
    }

    public void run()
    {
        RObject robject;
        RObject robject1;
        try
        {
            robject = Naming.getStub(1);
            robject1 = Naming.getStub(2);
        }
        catch(CORBException corbexception)
        {
            corbexception.printStackTrace();
            return;
        }
        do
            try
            {
                Thread.yield();
                Socket socket = serverSock.accept();
                log.println(4, "New connection from {0}", socket);
                Connection connection = new Connection(socket);
                ObjectOutputStream objectoutputstream = connection.getOutputStream();
                log.println(3, "Sending address to {0}", socket);
                InetAddress inetaddress = socket.getInetAddress();
                IpAddress ipaddress = new IpAddress(inetaddress.getAddress());
                objectoutputstream.writeObject(ipaddress);
                log.println(3, "Sending stubs to {0}", socket);
                objectoutputstream.writeObject(robject);
                objectoutputstream.writeObject(robject1);
                log.println(3, "Finished sending stubs to {0}", socket);
                ObjectInputStream objectinputstream = connection.getInputStream();
                int i = ((Integer)objectinputstream.readObject()).intValue();
                log.println(4, "Port number read " + i);
                log.println(3, "Getting lookup stubs from {0}", objectinputstream);
                RObject robject2 = (RObject)objectinputstream.readObject();
                RObject robject3 = (RObject)objectinputstream.readObject();
                log.println(3, "Got stubs from {0}", socket);
                OrbMonitor orbmonitor = new OrbMonitor(i, connection, requestQ);
                orbmonitor.setName("OrbMonitor" + connection);
                connection.setOrbMonitor(orbmonitor);
                String s = ipaddress + ":" + i;
                Naming.addStub(s, connection, robject2);
                Naming.addStub(s, connection, robject3);
                orbmonitor.start();
            }
            catch(Exception exception)
            {
                exception.printStackTrace();
            }
        while(true);
    }

    public static synchronized int getRequestId()
    {
        reqId++;
        return reqId;
    }

    public static Result callMethod(Request request, long l)
    {
        if(!started)
            throw new CORBRuntimeException("Orb has not been started");
        Result result = null;
        ServerObjectId serverobjectid = request.getServerId();
        int i = request.getRequestId();
        Connection connection = Naming.getConnection(serverobjectid);
        if(connection == null)
            return callLocalMethod(request);
        OrbMonitor orbmonitor = connection.getOrbMonitor();
        if(orbmonitor == null)
        {
            result = new Result(i);
            result.setStatus(2);
            result.setResult(new CORBException("No servicing thread for " + connection));
            result.flush();
            return result;
        }
        try
        {
            ObjectOutputStream objectoutputstream = connection.getOutputStream();
            ReplyQ replyq = new ReplyQ(log, i, l);
            log.println(5, "Calling method {0}", request);
            if(request.isOneWay())
            {
                synchronized(objectoutputstream)
                {
                    objectoutputstream.writeObject(request);
                    objectoutputstream.flush();
                    objectoutputstream.reset();
                }
            } else
            {
                orbmonitor.registerQueue(i, replyq);
                synchronized(objectoutputstream)
                {
                    objectoutputstream.writeObject(request);
                    objectoutputstream.flush();
                    objectoutputstream.reset();
                }
                log.println(5, "Waiting for Result for method {0}", request);
                result = replyq.dequeue();
                log.println(5, "Got Result for method {0}", request);
                orbmonitor.unregisterQueue(i);
            }
            return result;
        }
        catch(Throwable throwable)
        {
            log.println(1, "Error while calling method {0}", request);
            throwable.printStackTrace();
            Result result1 = new Result(i);
            result1.setStatus(2);
            result1.setResult(throwable);
            result1.flush();
            orbmonitor.unregisterQueue(i);
            return result1;
        }
    }

    private static Result callLocalMethod(Request request)
    {
        Result result = null;
        ServerObjectId serverobjectid = request.getServerId();
        int i = request.getRequestId();
        Object obj = null;
        IpAddress ipaddress = new IpAddress(serverobjectid.getIpAddress());
        if(serverobjectid.getPortNumber() == getPort() && ipaddress.equals(ipAddr))
            try
            {
                CORBSkeleton corbskeleton = (CORBSkeleton)Naming.getSkeleton(serverobjectid.getServerId());
                if(corbskeleton != null)
                    try
                    {
                        request.flush();
                        result = corbskeleton.dispatch(request);
                        if(request.isOneWay())
                        {
                            return null;
                        } else
                        {
                            result.flush();
                            return result;
                        }
                    }
                    catch(IOException ioexception)
                    {
                        log.println(1, "Could not flush request {0}", ioexception.getMessage());
                        result = new Result(i);
                        result.setStatus(2);
                        result.setResult(new CORBException(ioexception.getMessage()));
                        result.flush();
                        return result;
                    }
            }
            catch(Throwable throwable)
            {
                throwable.printStackTrace();
                log.println(1, "Could not locate skeleton {0}", throwable.getMessage());
                result = new Result(i);
                result.setStatus(2);
                result.setResult(throwable);
                result.flush();
                return result;
            }
        result = new Result(i);
        result.setStatus(2);
        result.setResult(new CORBException("Connection not found"));
        result.flush();
        return result;
    }

    static boolean started;
    static ServerSocket serverSock;
    static int reqId;
    static Log log = new Log("CORB", 1);
    static RequestQ requestQ;
    static IpAddress ipAddr;
    static int localPort;
    static HeartBeatThread heartBeatThread;
    static GarbageCollector garbageCollector;
    static Thread orbThread;

}