rounded up a few things for issue #40

src/Connections/CassandraConnectionException.cs

@@ -0,0 +1,16 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+
+namespace FluentCassandra.Connections
+{
+	public class CassandraConnectionException : CassandraException
+	{
+		public CassandraConnectionException(string message)
+			: base(message) { }
+
+		public CassandraConnectionException(string message, Exception innerException)
+			: base(message, innerException) { }
+	}
+}

src/Connections/Connection.cs

@@ -12,35 +12,49 @@ namespace FluentCassandra.Connections
 	/// <see href="http://github.com/robconery/NoRM/tree/master/NoRM/Connections/"/>
 	public class Connection : IConnection, IDisposable
 	{
-		private readonly ConnectionType _connectionType;
-		private readonly int _bufferSize;
-
-		private readonly TTransport _transport;
-		private readonly TProtocol _protocol;
-		private readonly Cassandra.Client _client;
+		private TTransport _transport;
+		private Cassandra.Client _client;
+		private readonly object _lock = new object();
 
 		/// <summary>
 		/// 
 		/// </summary>
+		/// <param name="server"></param>
 		/// <param name="builder"></param>
-		internal Connection(Server server, IConnectionBuilder builder)
-		{
-			_connectionType = builder.ConnectionType;
-			_bufferSize = builder.BufferSize;
+		public Connection(Server server, IConnectionBuilder builder)
+			: this(server, builder.ConnectionType, builder.BufferSize) { }
 
+		/// <summary>
+		/// 
+		/// </summary>
+		/// <param name="server"></param>
+		/// <param name="connectionType"></param>
+		/// <param name="bufferSize"></param>
+		public Connection(Server server, ConnectionType connectionType, int bufferSize)
+		{
 			Created = DateTime.UtcNow;
 			Server = server;
+			ConnectionType = connectionType;
+			BufferSize = bufferSize;
 
-			var socket = new TSocket(server.Host, server.Port, server.Timeout * 1000);
+			InitTransportAndClient();
+		}
+
+		/// <summary>
+		/// 
+		/// </summary>
+		private void InitTransportAndClient()
+		{
+			var socket = new TSocket(Server.Host, Server.Port, Server.Timeout * 1000);
 
-			switch (_connectionType)
+			switch (ConnectionType)
 			{
 				case ConnectionType.Simple:
 					_transport = socket;
 					break;
 
 				case ConnectionType.Buffered:
-					_transport = new TBufferedTransport(socket, _bufferSize);
+					_transport = new TBufferedTransport(socket, BufferSize);
 					break;
 
 				case ConnectionType.Framed:
@@ -51,8 +65,8 @@ internal Connection(Server server, IConnectionBuilder builder)
 					goto case ConnectionType.Framed;
 			}
 
-			_protocol = new TBinaryProtocol(_transport);
-			_client = new Cassandra.Client(_protocol);
+			var protocol = new TBinaryProtocol(_transport);
+			_client = new Cassandra.Client(protocol);
 		}
 
 		/// <summary>
@@ -73,6 +87,24 @@ public Server Server
 			private set;
 		}
 
+		/// <summary>
+		/// 
+		/// </summary>
+		public ConnectionType ConnectionType
+		{
+			get;
+			private set;
+		}
+
+		/// <summary>
+		/// 
+		/// </summary>
+		public int BufferSize
+		{
+			get;
+			private set;
+		}
+
 		/// <summary>
 		/// 
 		/// </summary>
@@ -83,11 +115,8 @@ public bool IsOpen
 				if (_transport == null)
 					return false;
 
-				lock (_transport)
-				{
-					try { return _transport.IsOpen; }
-					catch { return false; }
-				}
+				lock (_lock)
+					return _transport.IsOpen;
 			}
 		}
 
@@ -101,7 +130,10 @@ public void Open()
 			if (IsOpen)
 				return;
 
-			lock (_transport)
+			if (_transport == null)
+				InitTransportAndClient();
+
+			lock (_lock)
 				_transport.Open();
 		}
 
@@ -115,8 +147,12 @@ public void Close()
 			if (!IsOpen)
 				return;
 
-			lock (_transport)
+			lock (_lock)
+			{
 				_transport.Close();
+				_transport = null;
+				_client = null;
+			}
 		}
 
 		/// <summary>
@@ -125,6 +161,11 @@ public void Close()
 		/// <param name="keyspace"></param>
 		public void SetKeyspace(string keyspace)
 		{
+			CheckWasDisposed();
+
+			if (!IsOpen)
+				throw new CassandraConnectionException("A connection to Cassandra has not been opened.");
+
 			Client.set_keyspace(keyspace);
 		}
 
@@ -134,6 +175,11 @@ public void SetKeyspace(string keyspace)
 		/// <param name="cqlVersion"></param>
 		public void SetCqlVersion(string cqlVersion)
 		{
+			CheckWasDisposed();
+
+			if (!IsOpen)
+				throw new CassandraConnectionException("A connection to Cassandra has not been opened.");
+
 			Client.set_cql_version(cqlVersion);
 		}
 
@@ -142,10 +188,10 @@ public void SetCqlVersion(string cqlVersion)
 		/// </summary>
 		public Cassandra.Client Client
 		{
-			get 
-			{ 
-				CheckWasDisposed();
-				return _client; 
+			get
+			{
+				lock(_lock)
+					return _client;
 			}
 		}
 
@@ -195,10 +241,6 @@ protected virtual void Dispose(bool disposing)
 			if (!WasDisposed && disposing && _transport != null)
 			{
 				Close();
-
-				_client = null;
-				_protocol = null;
-				_transport = null;
 			}
 
 			WasDisposed = true;

src/FluentCassandra.csproj

@@ -106,6 +106,7 @@
     <Compile Include="CassandraSuperColumnFamilyOperations.cs" />
     <Compile Include="CassandraSuperColumnFamily.cs" />
     <Compile Include="CassandraTokenRange.cs" />
+    <Compile Include="Connections\CassandraConnectionException.cs" />
     <Compile Include="Connections\Connection.cs" />
     <Compile Include="Connections\ConnectionBuilder.cs" />
     <Compile Include="Connections\ConnectionProvider.cs" />

test/FluentCassandra.Tests/FluentCassandra.Tests.csproj

@@ -53,7 +53,7 @@
     <Compile Include="Bugs\Issue28GuidGeneratorInParallelContext.cs" />
     <Compile Include="Bugs\Issue25JavaBigDecimalBinaryConversion.cs" />
     <Compile Include="Bugs\Issue36KeyAliasSupport.cs" />
-    <Compile Include="Bugs\IssueXXCompositeTypeAsKey.cs" />
+    <Compile Include="Bugs\Issue39CompositeTypeAsKey.cs" />
     <Compile Include="CassandraDatabaseSetupFixture.cs" />
     <Compile Include="CassandraQueryTest.cs" />
     <Compile Include="Connections\ConnectionProviderTests.cs" />