---

TcpIpProtocol.java

Go to the documentation of this file.

/***
 * Jonathan: an Open Distributed Processing Environment 
 * Copyright (C) 1999-2000 France Telecom R&D
 * 
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2 of the License, or (at your option) any later version.
 * 
 * This library 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
 * Lesser General Public License for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 * 
 * Release: 3.0
 *
 * Contact: jonathan@objectweb.org
 *
 * Author: Bruno Dumant
 * 
 */

package org.objectweb.jonathan.libs.protocols.tcpip;

import java.io.IOException;

import org.objectweb.jonathan.apis.kernel.Context;
import org.objectweb.jonathan.apis.kernel.InternalException;
import org.objectweb.jonathan.apis.kernel.JonathanException;
import org.objectweb.jonathan.apis.presentation.EndOfMessageException;
import org.objectweb.jonathan.apis.presentation.Marshaller;
import org.objectweb.jonathan.apis.presentation.MarshallerFactory;
import org.objectweb.jonathan.apis.protocols.CommunicationException;
import org.objectweb.jonathan.apis.protocols.Protocol;
import org.objectweb.jonathan.apis.protocols.ProtocolGraph;
import org.objectweb.jonathan.apis.protocols.ReplyInterface;
import org.objectweb.jonathan.apis.protocols.SessionIdentifier;
import org.objectweb.jonathan.apis.protocols.Session_High;
import org.objectweb.jonathan.apis.protocols.Session_Low;
import org.objectweb.jonathan.apis.protocols.ip.IpConnection;
import org.objectweb.jonathan.apis.protocols.ip.IpSession;
import org.objectweb.jonathan.apis.protocols.ip.IpSessionIdentifier;
import org.objectweb.jonathan.apis.protocols.ip.TcpIpConnectionMgr;
import org.objectweb.jonathan.apis.protocols.ip.TcpIpSrvConnectionFactory;
import org.objectweb.jonathan.apis.resources.Chunk;
import org.objectweb.jonathan.apis.resources.ChunkFactory;
import org.objectweb.jonathan.apis.resources.ChunkProvider;
import org.objectweb.jonathan.apis.resources.Scheduler;

final public class TcpIpProtocol implements Protocol {
   Scheduler scheduler;
   ChunkFactory chunk_factory;
   MarshallerFactory marshaller_factory;
   
   public boolean verbose;

   SrvSessionFactory[] srv_session_factories = new SrvSessionFactory[4];
   int num_srv_sessions = 0;

   TcpIpConnectionMgr connection_mgr;
   
   public TcpIpProtocol(Context _c, TcpIpConnectionMgr connection_mgr,
                        Scheduler scheduler, ChunkFactory chunk_factory,
                        MarshallerFactory mf)
      throws JonathanException {
         this.scheduler = scheduler;
         this.chunk_factory = chunk_factory;
         this.connection_mgr = connection_mgr;
         this.marshaller_factory = mf;

         verbose = false;
         Object prop = _c.getValue("/jonathan/tcpip/verbose",'/');
         if (prop!=Context.NO_VALUE){
            verbose = ((Boolean) prop).booleanValue();
         }
   }

   public final boolean isAnInvocationProtocol() {
      return false;
   }

   public final ProtocolGraph newProtocolGraph() {
      return new TcpIpProtocolGraph();
   }
   
   public final ProtocolGraph newProtocolGraph(int port) {
      return new TcpIpProtocolGraph(port);
   }   
   
   public final IpSessionIdentifier newSessionIdentifier(String host,
                                                         int port) {
      return new CltSessionIdentifier(host,port);
   }
   
   public boolean isLocal(IpSessionIdentifier tcp_session_id) {
      String hostname = 
         connection_mgr.getCanonicalHostName(tcp_session_id.hostname);
      tcp_session_id.hostname = hostname;
      return isLocal(hostname,tcp_session_id.port);
   }

   boolean isLocal(String host,int port) {
      synchronized(srv_session_factories) {
         for (int i = 0; i < num_srv_sessions; i++) {
            if (srv_session_factories[i].session_id.port == port &&
                srv_session_factories[i].session_id.hostname.equals(host)) {
               return true;
            }
         }
         return false;
      }
   }

   void remove(SrvSessionFactory factory) {
      int i = 0;
      while(i < num_srv_sessions && srv_session_factories[i] != factory) {
         i++;
      }
      if (i < num_srv_sessions) {
         num_srv_sessions--;
         System.arraycopy(srv_session_factories,i+1,
                          srv_session_factories,i,num_srv_sessions - i);
         srv_session_factories[num_srv_sessions] = null;
      }
   }

   void send(Marshaller message,IpConnection connection)
      throws IOException {
      Chunk first = message.getState(), portion = null;
      int size = 0, len;
      boolean onechunk = true;
      Chunk c = first;

      while (c != null) {
         len = c.top - c.offset;
         if (len != 0) {
            if (size != 0) {
               onechunk = false;
            } else {
               portion = c;
            }
            size += len;
         }
         c = c.next;
      }
      if (onechunk) {
         if (portion != null) {
//               String str = "TCP a: ";
//               for (int i = 0; i <  portion.top - portion.offset; i++) {
//                  str = str + portion.data[i] + " ";
//               }
//               System.out.println(str);
            connection.emit(portion);
         }
         message.close();                  
      } else {
         try {
            portion = chunk_factory.newChunk(size);
            int off = portion.offset;
            c = first;
            while (c != null) {
               if ((len = c.top - c.offset) > 0) {
                  System.arraycopy(c.data,c.offset,portion.data,off,len);
                  off += len;
               }
               c = c.next; 
            }
            portion.top = off;
//             String str = "TCP b: ";            
//              for (int i = 0; i < size; i++) {
//                 str = portion.data[i] + " ";
//              }
//              System.out.println(str);
            connection.emit(portion);
            message.close();
         } finally {
            portion.release();
         }
      }
   }

   final class TcpIpProtocolGraph implements ProtocolGraph {

      int port;
      
      TcpIpProtocolGraph(int port) {
         this.port = port;
      }
      
      TcpIpProtocolGraph() {
         this(0);
      }

      public SessionIdentifier export(Session_Low hls)
         throws JonathanException {
         SrvSessionId session_id;
         synchronized (srv_session_factories) {
            // try to reuse an existing server socket factory.
            for (int i = 0; i < num_srv_sessions; i++) {
               if ((session_id = 
                    srv_session_factories[i].register(hls,this)) != null) {
                  return session_id;
               }
            }
            // it has not been possible to reuse an existing server socket
            // factory. create a new session identifier and srv session factory.
            TcpIpSrvConnectionFactory srv_connection_factory = 
               connection_mgr.newSrvConnectionFactory(port);
            session_id = new SrvSessionId(srv_connection_factory);
            SrvSessionFactory srv_fac =
               new SrvSessionFactory(session_id,hls,TcpIpProtocol.this);
            scheduler.newJob().run(srv_fac);
            int len = srv_session_factories.length;
            if (num_srv_sessions == len) {
               SrvSessionFactory[] new_srv_session_factories =
                  new SrvSessionFactory[len + 4];
               System.arraycopy(srv_session_factories,0,new_srv_session_factories,
                                0,len);
               srv_session_factories = new_srv_session_factories;
            }
            srv_session_factories[num_srv_sessions++] = srv_fac;
            return session_id;
         }
      }
   }   

   final class CltSessionIdentifier extends IpSessionIdentifier {
      
      CltSessionIdentifier(String hostname, int port) {
         super(hostname,port);
      }

      public Protocol getProtocol() {
         return TcpIpProtocol.this;
      }

      public void unexport() {}
      
      public Session_High bind(Session_Low hls)
         throws JonathanException {
          CltSession session = new CltSession(hls,this);
          IpConnection connection =
             connection_mgr.newCltConnection(hostname,port,session);
          session = (CltSession) connection.getSession();
          session.acquire();
          session.connect(connection);
          return session;
      }
   }
   
   final class SrvSessionId extends IpSessionIdentifier {
      
      TcpIpSrvConnectionFactory connection_factory;

      SrvSessionId(TcpIpSrvConnectionFactory connection_factory) {
         super(connection_factory.getHostName(),
               connection_factory.getPort());
         this.connection_factory = connection_factory;
      }
      
      public Protocol getProtocol() {
         return TcpIpProtocol.this;
      }

      public void unexport() {
         connection_factory.close();
      }
      
      public Session_High bind(Session_Low hls)
         throws JonathanException {
         throw new InternalException("Meaningless operation");
      }
   }
   
   abstract class Session implements IpSession, Runnable {
      IpConnection connection;
      TcpIpChunkProvider tcp_message;
      Session_Low hls;   

      Session(Session_Low hls) {
         super();
         this.hls = hls;
      }

      final public void prepare(Marshaller m) {}
      
      final public ReplyInterface prepareInvocation(Marshaller m)
         throws JonathanException {
            throw new InternalException("TCP session don't handle invocations.");
      }

      final public boolean direct() {
         return true;
      }
      
      public final Session_Low getHls() {
         return hls;
      }

      public synchronized void connect(IpConnection connection)
         throws JonathanException {
         if (this.connection != null) {
            return;
         }
         this.connection = connection;
         if (tcp_message == null) {
            // If tcp_message != null, a thread should already be started and 
            // running.
            tcp_message = new TcpIpChunkProvider(this);
            scheduler.newJob().run(this);
         }
      }
   

      public final IpConnection getConnection() {
         return connection;
      }
   
      final public void run() {
         TcpIpChunkProvider message = null;
         IpConnection connection = null;
         try {
            synchronized (this) {
               if (this.connection != null) {
                  message = tcp_message;
                  connection = this.connection;
               } else {
                  return;
               }
            }
            hls.send(marshaller_factory.newUnMarshaller(message),this);
         } catch (JonathanException e) {
            if (verbose) {
               System.err.println("Exception caught by TcpIpProtocol.");
            }
            Throwable f = e.represents();
            if (f instanceof IOException || f instanceof EndOfMessageException) {
               synchronized (this) {
                  if (connection == this.connection) {
                     unbind();
                     return;
                  } else {
                     message.delete();
                  }
               }
            }
            hls.send(e,this);
         }
      }

      final synchronized void closeNotify(TcpIpChunkProvider message) {
         if (connection != null && message == tcp_message) {
            tcp_message = new TcpIpChunkProvider(message);
            scheduler.newJob().run(this);
         } else {
            message.delete();
         }
      }

       final synchronized void deleteNotify(TcpIpChunkProvider message) {
           if (message == tcp_message) {
               tcp_message = null; // rebinding should work better now
           }
       }

      void unbind() {
         if (connection != null) {
            connection.delete();
            connection = null;
         }
      }

      TcpIpProtocol getProtocol() {
         return TcpIpProtocol.this;
      }
   }

   final class SrvSession extends Session {
      SrvSession(Session_Low hls) {
         super(hls);
      }

      final public void close() {}
      
      public final void send(Marshaller message)
         throws JonathanException {
            try {
               TcpIpProtocol.this.send(message,connection);
            } catch (IOException e) {
               try {
                  synchronized (this) {
                     if (connection == null) {
                        throw new CommunicationException("Session is closed");
                     }
                     unbind();
                     throw new CommunicationException(e);
                  }
               } finally {
                  message.close();
               }     
            }
      }
   }

   final class CltSession extends Session {
      IpSessionIdentifier session_id;
      int acquired;


      CltSession(Session_Low hls,IpSessionIdentifier session_id)
         throws JonathanException {
         super(hls);
         this.session_id = session_id;
         acquired = 0;
      }
   
      public final void send(Marshaller message)
         throws JonathanException {
         try {
            TcpIpProtocol.this.send(message,connection);
         } catch (IOException e) {
            try {
               synchronized (this) {
                  if (connection == null) {
                     throw new CommunicationException("Session is closed");
                  }
                  try {
                     rebind();
                  } catch (JonathanException g) {
                     throw new org.objectweb.jonathan.apis.binding.BindException("Can't rebind TCP session");
                  }
                  try {
                     TcpIpProtocol.this.send(message,connection);
                  } catch (IOException f) {
                     unbind();
                     throw new CommunicationException(f);
                  }
               }
            } finally {
               message.close();
            }
         }
      }
      
      void rebind() throws JonathanException {
         connection.delete();
         connection = null;
         String hostname = session_id.hostname;
         int port = session_id.port;
         IpConnection new_connection = 
            connection_mgr.newCltConnection(hostname,port,this);
         connect(new_connection);
      }

      final public synchronized void close() {
         acquired--;
         if (acquired == 0 && connection != null) {
            connection.release();
            connection = null;
         }
      }

      final synchronized void acquire() {
         acquired++;
      }
      
      public boolean equals(Object object) {
         if (object instanceof CltSession) {
            CltSession other = (CltSession) object;
            return other.session_id.equals(session_id) && other.hls.equals(hls);
         }
         return false;
      }

      public int hashCode() {
         return session_id.hashCode() + hls.hashCode();
      }
   }

   final class SrvSessionFactory implements Runnable {
      SrvSessionId session_id;
      Session_Low hls;
      boolean cont;
      TcpIpProtocol protocol;
      Thread runner;
   
      SrvSessionFactory(SrvSessionId session_id, Session_Low hls,
                        TcpIpProtocol protocol) {
         this.hls = hls;
         this.session_id = session_id;
         this.protocol = protocol;
         cont = true;
      }
   
      SrvSessionId register(Session_Low hls,
                            TcpIpProtocolGraph protocol_graph) {
         int port = protocol_graph.port;
         if ((port == 0 || port == session_id.port) && this.hls.equals(hls)) {
            return session_id;
         } else {
            return null;
         }
      }
   
      public void run() {
         runner = Thread.currentThread();
         TcpIpSrvConnectionFactory connection_factory = 
            session_id.connection_factory;
         while (cont) {
            try {
               SrvSession session = new  SrvSession(hls);
               IpConnection connection = 
                  connection_factory.newSrvConnection(session);
               session.connect(connection);
            } catch (JonathanException e) {
               if (cont) {
                  if (verbose) {
                     System.err.println("Stopping server socket on exception.");
                     e.printStackTrace();
                  }
                  session_id.unexport();
                  remove(this);
                  cont = false;
               }
            }
         }
      }
   
      void release() {
         cont = false;
         runner.interrupt();
         session_id.unexport();
         remove(this);
      }
   }   
}


final class TcpIpChunkProvider extends Chunk implements ChunkProvider  {
   static final byte[] empty_data = new byte[0];
   static final Chunk empty_chunk = new Chunk(empty_data,0,0);
   
   TcpIpProtocol.Session session;
   IpConnection connection;
   int max;

   Chunk cache;
   
   TcpIpChunkProvider(TcpIpProtocol.Session session) {
      super(empty_data,0,0);
      max = 0;
      this.session = session;
      connection = session.connection;
      cache = empty_chunk;
   }

   TcpIpChunkProvider(TcpIpChunkProvider message) {
      super(message.data,message.offset,message.top);
      cache = message.cache;
      max = message.max;
      session = message.session;
      connection = message.connection;
      message.cache = null;
   }
   
   public Chunk prepare() throws JonathanException {
      TcpIpProtocol protocol = session.getProtocol();
      if (top == offset) {
         // there is nothing left to read from this message
         try {
            int to_read = connection.available();
            if (to_read > 1) {
               if (to_read > max - top) {
                  cache.release();
                  cache = protocol.chunk_factory.newChunk(to_read);
                  data = cache.data;
                  offset = cache.offset;
                  top = cache.top;
                  max = data.length;
               }
               connection.receive(this,to_read);
               return this;
            } else {
               if (max - top == 0) {
                  cache.release();
                  cache = protocol.chunk_factory.newChunk();
                  data = cache.data;
                  offset = cache.offset;
                  top = cache.top;
                  max = data.length;
               }
               connection.receive(this,1);
               return this;
            }
         } catch (IOException e) {
            delete();
            connection.delete();
            connection = null;
            throw new JonathanException(e);
         } 
      } else {
         return this;
      }
   }

   public void close() {
      if (cache != null) {
         session.closeNotify(this);
      }
   }

   protected void finalize() {
      if (cache != null) {
         System.err.println("Resource management error. message " + this + 
                            " has not been properly closed.");
         delete();
      }
   }   

   final void delete() {
      if (cache != null) {
         cache.release();
         cache = null;
      }
      
      session.deleteNotify(this);
   }

   public Chunk duplicate() throws JonathanException {
      cache.top = top;
      return cache.duplicate(offset,top);
   }

   public Chunk duplicate(int off,int t) throws JonathanException {
      cache.top = top;
      return cache.duplicate(off,t);
   }
   
   public void release() {}   
}

---