CASSANDRA-241 optional support for framed transport

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@802898 13f79535-47bb-0310-9956-ffa450edef68

conf/storage-conf.xml

@@ -187,6 +187,11 @@
     <ThriftAddress>localhost</ThriftAddress>
     <!-- Thrift RPC port (the port clients connect to). -->
     <ThriftPort>9160</ThriftPort>
+    <!-- Whether or not to use a framed transport for Thrift. If this option
+         is set to true then you must also use a framed transport on the 
+         client-side, (framed and non-framed transports are not compatible).
+    -->
+    <ThriftFramedTransport>false</ThriftFramedTransport>
 
 
     <!--======================================================================-->

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -52,6 +52,7 @@ public class DatabaseDescriptor
     private static int storagePort_ = 7000;
     private static int controlPort_ = 7001;
     private static int thriftPort_ = 9160;
+    private static boolean thriftFramed_ = false;
     private static String listenAddress_; // leave null so we can fall through to getLocalHost
     private static String thriftAddress_;
     private static String clusterName_ = "Test";
@@ -263,6 +264,23 @@ public class DatabaseDescriptor
             if (port != null)
                 thriftPort_ = Integer.parseInt(port);
 
+            /* Framed (Thrift) transport (default to "no") */
+            String framedRaw = xmlUtils.getNodeValue("/Storage/ThriftFramedTransport");
+            if (framedRaw != null)
+            {
+                if (framedRaw.compareToIgnoreCase("true") == 0 || 
+                        framedRaw.compareToIgnoreCase("false") == 0)
+                {
+                    thriftFramed_ = Boolean.valueOf(framedRaw);
+                }
+                else
+                {
+                    throw new ConfigurationException("Unrecognized value " + 
+                            "for ThriftFramedTransport.  Use 'true' or 'false'."); 
+                }
+            }
+            
+
             /* Number of days to keep the memtable around w/o flushing */
             String lifetime = xmlUtils.getNodeValue("/Storage/MemtableLifetimeInDays");
             if ( lifetime != null )
@@ -524,6 +542,11 @@ public class DatabaseDescriptor
         }
     }
 
+    public static boolean isThriftFramed()
+    {
+        return thriftFramed_;
+    }
+
     private static AbstractType getComparator(Node columnFamily, String attr)
     throws ConfigurationException, TransformerException, NoSuchMethodException, InvocationTargetException, IllegalAccessException, InstantiationException
     {

src/java/org/apache/cassandra/service/CassandraDaemon.java

@@ -30,6 +30,7 @@ import org.apache.thrift.server.TThreadPoolServer;
 import org.apache.thrift.transport.TServerSocket;
 import org.apache.thrift.transport.TTransportException;
 import org.apache.thrift.transport.TTransportFactory;
+import org.apache.thrift.transport.TFramedTransport;
 import org.apache.thrift.TProcessorFactory;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.utils.FBUtilities;
@@ -107,14 +108,28 @@ public class CassandraDaemon
 
         // Protocol factory
         TProtocolFactory tProtocolFactory = new TBinaryProtocol.Factory();
+        
+        // Transport factory
+        TTransportFactory inTransportFactory, outTransportFactory;
+        if (DatabaseDescriptor.isThriftFramed())
+        {
+            inTransportFactory = new TFramedTransport.Factory();
+            outTransportFactory = new TFramedTransport.Factory();
+            
+        }
+        else
+        {
+            inTransportFactory = new TTransportFactory();
+            outTransportFactory = new TTransportFactory();
+        }
 
         // ThreadPool Server
         TThreadPoolServer.Options options = new TThreadPoolServer.Options();
         options.minWorkerThreads = 64;
         serverEngine = new TThreadPoolServer(new TProcessorFactory(processor),
                                              tServerSocket,
-                                             new TTransportFactory(),
-                                             new TTransportFactory(),
+                                             inTransportFactory,
+                                             outTransportFactory,
                                              tProtocolFactory,
                                              tProtocolFactory,
                                              options);