CPP-811 Add unit tests for request handler changes (#302)

Author: mpenick

cpp-driver/gtests/src/unit/tests/test_pool.cpp

@@ -70,9 +70,9 @@ class PoolUnitTest : public LoopTest {
       : public RequestState
       , public Status<RequestState::Enum> {
   public:
-    RequestStatus(uv_loop_t* loop, int num_nodes = NUM_NODES)
+    RequestStatus(uv_loop_t* loop, int num_requests = NUM_NODES)
         : loop_(loop)
-        , remaining_(num_nodes) {}
+        , remaining_(num_requests) {}
 
     virtual void set(RequestState::Enum state) {
       Status<RequestStatus::Enum>::set(state);
@@ -88,13 +88,13 @@ class PoolUnitTest : public LoopTest {
 
   protected:
     uv_loop_t* loop_;
-    size_t remaining_;
+    int remaining_;
   };
 
   class RequestStatusWithManager : public RequestStatus {
   public:
-    RequestStatusWithManager(uv_loop_t* loop, int num_nodes = NUM_NODES)
-        : RequestStatus(loop, num_nodes) {}
+    RequestStatusWithManager(uv_loop_t* loop, int num_requests = NUM_NODES)
+        : RequestStatus(loop, num_requests) {}
 
     ~RequestStatusWithManager() {
       ConnectionPoolManager::Ptr temp(manager());
@@ -354,6 +354,33 @@ class PoolUnitTest : public LoopTest {
     }
   }
 
+  static void on_pool_connected_exhaust_streams(ConnectionPoolManagerInitializer* initializer,
+                                                RequestStatusWithManager* status) {
+    const Address address("127.0.0.1", 9042);
+    ConnectionPoolManager::Ptr manager = initializer->release_manager();
+    status->set_manager(manager);
+
+    for (size_t i = 0; i < CASS_MAX_STREAMS; ++i) {
+      PooledConnection::Ptr connection = manager->find_least_busy(address);
+
+      if (connection) {
+        RequestCallback::Ptr callback(new RequestCallback(status));
+        if (connection->write(callback.get()) < 0) {
+          status->error_failed_write();
+        }
+      } else {
+        status->error_no_connection();
+      }
+    }
+
+    PooledConnection::Ptr connection = manager->find_least_busy(address);
+    ASSERT_TRUE(connection);
+    RequestCallback::Ptr callback(new RequestCallback(status));
+    EXPECT_EQ(connection->write(callback.get()), Request::REQUEST_ERROR_NO_AVAILABLE_STREAM_IDS);
+
+    manager->flush();
+  }
+
   static void on_pool_nop(ConnectionPoolManagerInitializer* initializer,
                           RequestStatusWithManager* status) {
     ConnectionPoolManager::Ptr manager = initializer->release_manager();
@@ -798,6 +825,17 @@ TEST_F(PoolUnitTest, PartialReconnect) {
   // TODO:
 }
 
-TEST_F(PoolUnitTest, LowNumberOfStreams) {
-  // TODO:
+TEST_F(PoolUnitTest, NoAvailableStreams) {
+  mockssandra::SimpleCluster cluster(simple(), 1);
+  ASSERT_EQ(cluster.start_all(), 0);
+
+  RequestStatusWithManager status(loop(), CASS_MAX_STREAMS);
+
+  ConnectionPoolManagerInitializer::Ptr initializer(new ConnectionPoolManagerInitializer(
+      PROTOCOL_VERSION, bind_callback(on_pool_connected_exhaust_streams, &status)));
+
+  initializer->initialize(loop(), hosts());
+  uv_run(loop(), UV_RUN_DEFAULT);
+
+  EXPECT_EQ(status.count(RequestStatus::SUCCESS), CASS_MAX_STREAMS) << status.results();
 }

cpp-driver/gtests/src/unit/tests/test_request_processor.cpp

@@ -27,19 +27,91 @@
 using namespace datastax::internal;
 using namespace datastax::internal::core;
 
+class InorderLoadBalancingPolicy : public LoadBalancingPolicy {
+public:
+  typedef SharedRefPtr<LoadBalancingPolicy> Ptr;
+  typedef Vector<Ptr> Vec;
+
+  InorderLoadBalancingPolicy()
+      : LoadBalancingPolicy()
+      , hosts_(new HostVec()) {}
+
+  virtual void init(const Host::Ptr& connected_host, const HostMap& hosts, Random* random,
+                    const String& local_dc) {
+    hosts_->reserve(hosts.size());
+    std::transform(hosts.begin(), hosts.end(), std::back_inserter(*hosts_), GetHost());
+  }
+
+  virtual CassHostDistance distance(const Host::Ptr& host) const {
+    return CASS_HOST_DISTANCE_LOCAL;
+  }
+
+  virtual bool is_host_up(const Address& address) const {
+    return std::find_if(hosts_->begin(), hosts_->end(), FindAddress(address)) != hosts_->end();
+  }
+
+  virtual void on_host_added(const Host::Ptr& host) { add_host(hosts_, host); }
+
+  virtual void on_host_removed(const Host::Ptr& host) { remove_host(hosts_, host); }
+
+  virtual void on_host_up(const Host::Ptr& host) { add_host(hosts_, host); }
+
+  virtual void on_host_down(const Address& address) { remove_host(hosts_, address); }
+
+  virtual QueryPlan* new_query_plan(const String& keyspace, RequestHandler* request_handler,
+                                    const TokenMap* token_map) {
+    return new InternalQueryPlan(hosts_);
+  }
+
+  virtual LoadBalancingPolicy* new_instance() { return new InorderLoadBalancingPolicy(); }
+
+private:
+  struct FindAddress {
+
+    FindAddress(const Address& address)
+        : address(address) {}
+
+    bool operator()(const Host::Ptr& host) const { return host->address() == address; }
+
+    Address address;
+  };
+
+  class InternalQueryPlan : public datastax::internal::core::QueryPlan {
+  public:
+    InternalQueryPlan(const CopyOnWriteHostVec& hosts)
+        : index_(0)
+        , hosts_(hosts) {}
+
+    virtual Host::Ptr compute_next() {
+      if (index_ < hosts_->size()) {
+        return (*hosts_)[index_++];
+      }
+      return Host::Ptr();
+    }
+
+  private:
+    size_t index_;
+    CopyOnWriteHostVec hosts_;
+  };
+
+private:
+  CopyOnWriteHostVec hosts_;
+};
+
 class RequestProcessorUnitTest : public EventLoopTest {
 public:
   RequestProcessorUnitTest()
       : EventLoopTest("RequestProcessorUnitTest") {}
 
-  HostMap generate_hosts() {
+  HostMap generate_hosts(size_t num_hosts = 3) {
     HostMap hosts;
-    Host::Ptr host1(new Host(Address("127.0.0.1", PORT)));
-    Host::Ptr host2(new Host(Address("127.0.0.2", PORT)));
-    Host::Ptr host3(new Host(Address("127.0.0.3", PORT)));
-    hosts[host1->address()] = host1;
-    hosts[host2->address()] = host2;
-    hosts[host3->address()] = host3;
+    num_hosts = std::min(num_hosts, static_cast<size_t>(255));
+    for (size_t i = 1; i <= num_hosts; ++i) {
+      char buf[64];
+      sprintf(buf, "127.0.0.%d", static_cast<int>(i));
+      Host::Ptr host(new Host(Address(buf, PORT)));
+      hosts[host->address()] = host;
+    }
     return hosts;
   }
 
@@ -619,3 +691,73 @@ TEST_F(RequestProcessorUnitTest, RequestTimeout) {
   processor->close();
   ASSERT_TRUE(close_future->wait_for(WAIT_FOR_TIME));
 }
+
+TEST_F(RequestProcessorUnitTest, LowNumberOfStreams) {
+  mockssandra::SimpleRequestHandlerBuilder builder;
+  builder.on(mockssandra::OPCODE_QUERY)
+      .wait(1000) // Give  time for the streams to run out
+      .system_local()
+      .system_peers()
+      .empty_rows_result(1);
+  mockssandra::SimpleCluster cluster(builder.build(), 2); // Two node cluster
+  ASSERT_EQ(cluster.start_all(), 0);
+
+  Future::Ptr close_future(new Future());
+  CloseListener::Ptr listener(new CloseListener(close_future));
+
+  HostMap hosts(generate_hosts(2));
+  Future::Ptr connect_future(new Future());
+
+  ExecutionProfile profile;
+  profile.set_load_balancing_policy(new InorderLoadBalancingPolicy());
+  profile.set_speculative_execution_policy(new NoSpeculativeExecutionPolicy());
+  profile.set_retry_policy(new DefaultRetryPolicy());
+
+  RequestProcessorSettings settings;
+  settings.default_profile = profile;
+  settings.request_queue_size = 2 * CASS_MAX_STREAMS + 1; // Create a request queue with enough room
+
+  RequestProcessorInitializer::Ptr initializer(new RequestProcessorInitializer(
+      hosts.begin()->second, PROTOCOL_VERSION, hosts, TokenMap::Ptr(), "",
+      bind_callback(on_connected, connect_future.get())));
+  initializer->with_settings(settings)->with_listener(listener.get())->initialize(event_loop());
+
+  ASSERT_TRUE(connect_future->wait_for(WAIT_FOR_TIME));
+  EXPECT_FALSE(connect_future->error());
+  RequestProcessor::Ptr processor(connect_future->processor());
+
+  // Saturate the hosts connections, but leave one stream.
+  for (int i = 0; i < 2 * CASS_MAX_STREAMS - 1; ++i) {
+    ResponseFuture::Ptr response_future(new ResponseFuture());
+    Statement::Ptr request(new QueryRequest("SELECT * FROM table"));
+    RequestHandler::Ptr request_handler(new RequestHandler(request, response_future));
+    processor->process_request(request_handler);
+  }
+
+  { // Try two more requests. One should succeed on "127.0.0.2" and the other should fail (out of
+    // streams).
+    ResponseFuture::Ptr response_future(new ResponseFuture());
+
+    Statement::Ptr request(new QueryRequest("SELECT * FROM table"));
+    request->set_record_attempted_addresses(true);
+    RequestHandler::Ptr request_handler(new RequestHandler(request, response_future));
+    processor->process_request(request_handler);
+
+    ResponseFuture::Ptr response_future_fail(new ResponseFuture());
+    RequestHandler::Ptr request_handler_fail(new RequestHandler(
+        Statement::Ptr(new QueryRequest("SELECT * FROM table")), response_future_fail));
+    processor->process_request(request_handler_fail);
+    ASSERT_TRUE(response_future_fail->wait_for(WAIT_FOR_TIME));
+    ASSERT_TRUE(response_future_fail->error());
+    EXPECT_EQ(CASS_ERROR_LIB_NO_HOSTS_AVAILABLE, response_future_fail->error()->code);
+
+    ASSERT_TRUE(response_future->wait_for(WAIT_FOR_TIME));
+    EXPECT_FALSE(response_future->error());
+    AddressVec attempted = response_future->attempted_addresses();
+    ASSERT_GE(attempted.size(), 1u);
+    EXPECT_EQ(attempted[0], Address("127.0.0.2", PORT));
+  }
+
+  processor->close();
+  ASSERT_TRUE(close_future->wait_for(WAIT_FOR_TIME));
+}

cpp-driver/gtests/src/unit/tests/test_session_base.cpp

@@ -218,6 +218,25 @@ TEST_F(SessionBaseUnitTest, InvalidProtocol) {
   EXPECT_EQ(0, session_base.closed());
 }
 
+TEST_F(SessionBaseUnitTest, UnsupportedProtocol) {
+  mockssandra::SimpleCluster cluster(simple());
+  ASSERT_EQ(cluster.start_all(), 0);
+
+  Config config;
+  config.set_protocol_version(ProtocolVersion(2)); // Unsupported protocol version
+  config.contact_points().push_back(Address("127.0.0.1", 9042));
+  TestSessionBase session_base;
+
+  Future::Ptr connect_future(session_base.connect(config, KEYSPACE));
+  ASSERT_TRUE(connect_future->wait_for(WAIT_FOR_TIME));
+  EXPECT_EQ(CASS_ERROR_LIB_NO_HOSTS_AVAILABLE, connect_future->error()->code);
+  EXPECT_TRUE(connect_future->error()->message.find(
+                  "Operation unsupported by this protocol version") != String::npos);
+  EXPECT_EQ(0, session_base.connected());
+  EXPECT_EQ(1, session_base.failed());
+  EXPECT_EQ(0, session_base.closed());
+}
+
 TEST_F(SessionBaseUnitTest, SslError) {
   mockssandra::SimpleCluster cluster(simple());
   use_ssl(&cluster);

cpp-driver/gtests/src/unit/tests/test_statement.cpp

@@ -16,6 +16,7 @@
 
 #include "unit.hpp"
 
+#include "batch_request.hpp"
 #include "constants.hpp"
 #include "control_connection.hpp"
 #include "query_request.hpp"
@@ -123,3 +124,45 @@ TEST_F(StatementUnitTest, SetHostWhereHostIsDown) {
   ASSERT_TRUE(future->error());
   EXPECT_EQ(future->error()->code, CASS_ERROR_LIB_NO_HOSTS_AVAILABLE);
 }
+
+TEST_F(StatementUnitTest, ErrorBatchWithNamedParameters) {
+  mockssandra::SimpleCluster cluster(simple(), 1);
+  ASSERT_EQ(cluster.start_all(), 0);
+
+  connect();
+
+  BatchRequest::Ptr batch(new BatchRequest(CASS_BATCH_TYPE_UNLOGGED));
+
+  Statement::Ptr request(new QueryRequest("SELECT * FROM does_not_matter WHERE key = ?",
+                                          1)); // Space for a named parameter
+
+  request->set("key", 42); // Use named parameters
+
+  batch->add_statement(request.get());
+
+  ResponseFuture::Ptr future(session.execute(Request::ConstPtr(batch)));
+  future->wait();
+
+  ASSERT_TRUE(future->error());
+  EXPECT_EQ(future->error()->code, CASS_ERROR_LIB_BAD_PARAMS);
+  EXPECT_EQ(future->error()->message, "Batches cannot contain queries with named values");
+}
+
+TEST_F(StatementUnitTest, ErrorParametersUnset) {
+  mockssandra::SimpleCluster cluster(simple(), 1);
+  ASSERT_EQ(cluster.start_all(), 0);
+
+  Config config;
+  config.set_protocol_version(ProtocolVersion(3));
+
+  connect(config);
+
+  Statement::Ptr request(new QueryRequest("SELECT * FROM does_not_matter WHERE key = ?",
+                                          1)); // Parameters start as unset
+
+  ResponseFuture::Ptr future(session.execute(Request::ConstPtr(request)));
+  future->wait();
+
+  ASSERT_TRUE(future->error());
+  EXPECT_EQ(future->error()->code, CASS_ERROR_LIB_PARAMETER_UNSET);
+}

cpp-driver/src/batch_request.hpp

@@ -33,11 +33,12 @@ class ExecuteRequest;
 
 class BatchRequest : public RoutableRequest {
 public:
+  typedef SharedRefPtr<BatchRequest> Ptr;
   typedef Vector<Statement::Ptr> StatementVec;
 
-  BatchRequest(uint8_t type_)
+  BatchRequest(uint8_t type)
       : RoutableRequest(CQL_OPCODE_BATCH)
-      , type_(type_) {}
+      , type_(type) {}
 
   uint8_t type() const { return type_; }