codegen: Report DAQmx event registration failure by closing stream

generated/nidaqmx/nidaqmx_service.cpp

@@ -14020,15 +14020,15 @@ namespace nidaqmx_grpc {
     using CallbackRouter = nidevice_grpc::CallbackRouter<int32, TaskHandle, int32>;
     class RegisterDoneEventReactor : public nidevice_grpc::ServerWriterReactor<RegisterDoneEventResponse, nidevice_grpc::CallbackRegistration> {
     public:
-    RegisterDoneEventReactor(const RegisterDoneEventRequest& request, NiDAQmxLibraryInterface* library, const ResourceRepositorySharedPtr& session_repository)
+    RegisterDoneEventReactor(::grpc::CallbackServerContext* context, const RegisterDoneEventRequest* request, NiDAQmxLibraryInterface* library, NiDAQmxService* service)
     {
-      auto status = start(&request, library, session_repository);
+      auto status = start(context, request, library, service);
       if (!status.ok()) {
-        this->Finish(status);
+        this->try_finish(std::move(status));
       }
     }
 
-    ::grpc::Status start(const RegisterDoneEventRequest* request, NiDAQmxLibraryInterface* library, const ResourceRepositorySharedPtr& session_repository_)
+    ::grpc::Status start(::grpc::CallbackServerContext* context, const RegisterDoneEventRequest* request, NiDAQmxLibraryInterface* library, NiDAQmxService* service)
     {
       try {
         auto handler = CallbackRouter::register_handler(
@@ -14040,21 +14040,19 @@ namespace nidaqmx_grpc {
             return 0;
         });
 
+        const auto& session_repository_ = service->session_repository_;
         auto task_grpc_session = request->task();
         TaskHandle task = session_repository_->access_session(task_grpc_session.name());
         auto options = 0U;
 
         auto status = library->RegisterDoneEvent(task, options, CallbackRouter::handle_callback, handler->token());
+        if (!status_ok(status)) {
+          return service->ConvertApiErrorStatusForTaskHandle(context, status, task);
+        }
 
         // SendInitialMetadata after the driver call so that WaitForInitialMetadata can be used to ensure that calls are serialized.
         StartSendInitialMetadata();
 
-        if (status) {
-          RegisterDoneEventResponse failed_to_register_response;
-          failed_to_register_response.set_status(status);
-          queue_write(failed_to_register_response);
-        }
-
         this->set_producer(std::move(handler));
       }
       catch (nidevice_grpc::NonDriverException& ex) {
@@ -14065,7 +14063,7 @@ namespace nidaqmx_grpc {
     }
     };
 
-    return new RegisterDoneEventReactor(*request, library_, session_repository_);
+    return new RegisterDoneEventReactor(context, request, library_, this);
   }
 
   //---------------------------------------------------------------------
@@ -14076,15 +14074,15 @@ namespace nidaqmx_grpc {
     using CallbackRouter = nidevice_grpc::CallbackRouter<int32, TaskHandle, int32, uInt32>;
     class RegisterEveryNSamplesEventReactor : public nidevice_grpc::ServerWriterReactor<RegisterEveryNSamplesEventResponse, nidevice_grpc::CallbackRegistration> {
     public:
-    RegisterEveryNSamplesEventReactor(const RegisterEveryNSamplesEventRequest& request, NiDAQmxLibraryInterface* library, const ResourceRepositorySharedPtr& session_repository)
+    RegisterEveryNSamplesEventReactor(::grpc::CallbackServerContext* context, const RegisterEveryNSamplesEventRequest* request, NiDAQmxLibraryInterface* library, NiDAQmxService* service)
     {
-      auto status = start(&request, library, session_repository);
+      auto status = start(context, request, library, service);
       if (!status.ok()) {
-        this->Finish(status);
+        this->try_finish(std::move(status));
       }
     }
 
-    ::grpc::Status start(const RegisterEveryNSamplesEventRequest* request, NiDAQmxLibraryInterface* library, const ResourceRepositorySharedPtr& session_repository_)
+    ::grpc::Status start(::grpc::CallbackServerContext* context, const RegisterEveryNSamplesEventRequest* request, NiDAQmxLibraryInterface* library, NiDAQmxService* service)
     {
       try {
         auto handler = CallbackRouter::register_handler(
@@ -14098,6 +14096,7 @@ namespace nidaqmx_grpc {
             return 0;
         });
 
+        const auto& session_repository_ = service->session_repository_;
         auto task_grpc_session = request->task();
         TaskHandle task = session_repository_->access_session(task_grpc_session.name());
         int32 every_n_samples_event_type;
@@ -14120,16 +14119,13 @@ namespace nidaqmx_grpc {
         auto options = 0U;
 
         auto status = library->RegisterEveryNSamplesEvent(task, every_n_samples_event_type, n_samples, options, CallbackRouter::handle_callback, handler->token());
+        if (!status_ok(status)) {
+          return service->ConvertApiErrorStatusForTaskHandle(context, status, task);
+        }
 
         // SendInitialMetadata after the driver call so that WaitForInitialMetadata can be used to ensure that calls are serialized.
         StartSendInitialMetadata();
 
-        if (status) {
-          RegisterEveryNSamplesEventResponse failed_to_register_response;
-          failed_to_register_response.set_status(status);
-          queue_write(failed_to_register_response);
-        }
-
         this->set_producer(std::move(handler));
       }
       catch (nidevice_grpc::NonDriverException& ex) {
@@ -14140,7 +14136,7 @@ namespace nidaqmx_grpc {
     }
     };
 
-    return new RegisterEveryNSamplesEventReactor(*request, library_, session_repository_);
+    return new RegisterEveryNSamplesEventReactor(context, request, library_, this);
   }
 
   //---------------------------------------------------------------------
@@ -14151,15 +14147,15 @@ namespace nidaqmx_grpc {
     using CallbackRouter = nidevice_grpc::CallbackRouter<int32, TaskHandle, int32>;
     class RegisterSignalEventReactor : public nidevice_grpc::ServerWriterReactor<RegisterSignalEventResponse, nidevice_grpc::CallbackRegistration> {
     public:
-    RegisterSignalEventReactor(const RegisterSignalEventRequest& request, NiDAQmxLibraryInterface* library, const ResourceRepositorySharedPtr& session_repository)
+    RegisterSignalEventReactor(::grpc::CallbackServerContext* context, const RegisterSignalEventRequest* request, NiDAQmxLibraryInterface* library, NiDAQmxService* service)
     {
-      auto status = start(&request, library, session_repository);
+      auto status = start(context, request, library, service);
       if (!status.ok()) {
-        this->Finish(status);
+        this->try_finish(std::move(status));
       }
     }
 
-    ::grpc::Status start(const RegisterSignalEventRequest* request, NiDAQmxLibraryInterface* library, const ResourceRepositorySharedPtr& session_repository_)
+    ::grpc::Status start(::grpc::CallbackServerContext* context, const RegisterSignalEventRequest* request, NiDAQmxLibraryInterface* library, NiDAQmxService* service)
     {
       try {
         auto handler = CallbackRouter::register_handler(
@@ -14171,6 +14167,7 @@ namespace nidaqmx_grpc {
             return 0;
         });
 
+        const auto& session_repository_ = service->session_repository_;
         auto task_grpc_session = request->task();
         TaskHandle task = session_repository_->access_session(task_grpc_session.name());
         int32 signal_id;
@@ -14192,16 +14189,13 @@ namespace nidaqmx_grpc {
         auto options = 0U;
 
         auto status = library->RegisterSignalEvent(task, signal_id, options, CallbackRouter::handle_callback, handler->token());
+        if (!status_ok(status)) {
+          return service->ConvertApiErrorStatusForTaskHandle(context, status, task);
+        }
 
         // SendInitialMetadata after the driver call so that WaitForInitialMetadata can be used to ensure that calls are serialized.
         StartSendInitialMetadata();
 
-        if (status) {
-          RegisterSignalEventResponse failed_to_register_response;
-          failed_to_register_response.set_status(status);
-          queue_write(failed_to_register_response);
-        }
-
         this->set_producer(std::move(handler));
       }
       catch (nidevice_grpc::NonDriverException& ex) {
@@ -14212,7 +14206,7 @@ namespace nidaqmx_grpc {
     }
     };
 
-    return new RegisterSignalEventReactor(*request, library_, session_repository_);
+    return new RegisterSignalEventReactor(context, request, library_, this);
   }
 
   //---------------------------------------------------------------------

generated/nifake_non_ivi/nifake_non_ivi_service.cpp

@@ -770,15 +770,15 @@ namespace nifake_non_ivi_grpc {
     using CallbackRouter = nidevice_grpc::CallbackRouter<int32, myInt16>;
     class RegisterCallbackReactor : public nidevice_grpc::ServerWriterReactor<RegisterCallbackResponse, nidevice_grpc::CallbackRegistration> {
     public:
-    RegisterCallbackReactor(const RegisterCallbackRequest& request, NiFakeNonIviLibraryInterface* library, const ResourceRepositorySharedPtr& session_repository)
+    RegisterCallbackReactor(::grpc::CallbackServerContext* context, const RegisterCallbackRequest* request, NiFakeNonIviLibraryInterface* library, NiFakeNonIviService* service)
     {
-      auto status = start(&request, library, session_repository);
+      auto status = start(context, request, library, service);
       if (!status.ok()) {
-        this->Finish(status);
+        this->try_finish(std::move(status));
       }
     }
 
-    ::grpc::Status start(const RegisterCallbackRequest* request, NiFakeNonIviLibraryInterface* library, const ResourceRepositorySharedPtr& session_repository_)
+    ::grpc::Status start(::grpc::CallbackServerContext* context, const RegisterCallbackRequest* request, NiFakeNonIviLibraryInterface* library, NiFakeNonIviService* service)
     {
       try {
         auto handler = CallbackRouter::register_handler(
@@ -790,19 +790,17 @@ namespace nifake_non_ivi_grpc {
             return 0;
         });
 
+        const auto& session_repository_ = service->session_repository_;
         myInt16 input_data = request->input_data();
 
         auto status = library->RegisterCallback(input_data, CallbackRouter::handle_callback, handler->token());
+        if (!status_ok(status)) {
+          return service->ConvertApiErrorStatusForFakeHandle(context, status, 0);
+        }
 
         // SendInitialMetadata after the driver call so that WaitForInitialMetadata can be used to ensure that calls are serialized.
         StartSendInitialMetadata();
 
-        if (status) {
-          RegisterCallbackResponse failed_to_register_response;
-          failed_to_register_response.set_status(status);
-          queue_write(failed_to_register_response);
-        }
-
         this->set_producer(std::move(handler));
       }
       catch (nidevice_grpc::NonDriverException& ex) {
@@ -813,7 +811,7 @@ namespace nifake_non_ivi_grpc {
     }
     };
 
-    return new RegisterCallbackReactor(*request, library_, session_repository_);
+    return new RegisterCallbackReactor(context, request, library_, this);
   }
 
   //---------------------------------------------------------------------

source/codegen/templates/service_helpers.mako

@@ -168,19 +168,20 @@ ${populate_response(function_data=function_data, parameters=parameters, indent_l
   request_type = service_helpers.get_request_type(function_name)
   response_type = service_helpers.get_response_type(function_name)
   driver_library_interface = common_helpers.get_library_interface_type_name(config)
+  service_class = config["service_class_prefix"] + "Service"
 %>\
     using CallbackRouter = nidevice_grpc::CallbackRouter<int32, ${service_helpers.create_param_type_list(callback_parameters)}>;
     class ${function_name}Reactor : public nidevice_grpc::ServerWriterReactor<${response_type}, nidevice_grpc::CallbackRegistration> {
     public:
-    ${function_name}Reactor(const ${request_type}& request, ${driver_library_interface}* library, const ResourceRepositorySharedPtr& session_repository)
+    ${function_name}Reactor(::grpc::CallbackServerContext* context, const ${request_type}* request, ${driver_library_interface}* library, ${service_class}* service)
     {
-      auto status = start(&request, library, session_repository);
+      auto status = start(context, request, library, service);
       if (!status.ok()) {
-        this->Finish(status);
+        this->try_finish(std::move(status));
       }
     }
 
-    ::grpc::Status start(const ${request_type}* request, ${driver_library_interface}* library, const ResourceRepositorySharedPtr& session_repository_)
+    ::grpc::Status start(::grpc::CallbackServerContext* context, const ${request_type}* request, ${driver_library_interface}* library, ${service_class}* service)
     {
       try {
         auto handler = CallbackRouter::register_handler(
@@ -194,6 +195,7 @@ ${set_response_values(output_parameters=response_parameters, init_method=False)}
             return 0;
         });
 
+        const auto& session_repository_ = service->session_repository_;
 <%block filter="common_helpers.indent(1)">\
 ${initialize_input_params(function_name, parameters)}\
 </%block>\
@@ -205,16 +207,11 @@ ${call_library_method(
   library_lval="library",
   indent_level=1)
 }\
+${populate_error_check(function_data, parameters, indent_level=1, service_deref="service->")}\
 
         // SendInitialMetadata after the driver call so that WaitForInitialMetadata can be used to ensure that calls are serialized.
         StartSendInitialMetadata();
 
-        if (status) {
-          ${response_type} failed_to_register_response;
-          failed_to_register_response.set_status(status);
-          queue_write(failed_to_register_response);
-        }
-
         this->set_producer(std::move(handler));
       }
       catch (nidevice_grpc::NonDriverException& ex) {
@@ -225,7 +222,7 @@ ${call_library_method(
     }
     };
 
-    return new ${function_name}Reactor(*request, library_, session_repository_);
+    return new ${function_name}Reactor(context, request, library_, this);
 </%def>
 
 ## Generate the core method body for a method with repeated varargs.
@@ -751,7 +748,7 @@ ${set_response_values(normal_outputs, init_method)}\
 </%def>
 
 ## Handles populating the response message after calling the driver API.
-<%def name="populate_error_check(function_data, parameters, indent_level=0)">\
+<%def name="populate_error_check(function_data, parameters, indent_level=0, service_deref='')">\
 <%
   config = data['config']
   input_parameters = [p for p in parameters if common_helpers.is_input_parameter(p)]
@@ -772,7 +769,7 @@ ${set_response_values(normal_outputs, init_method)}\
   if function_data.get('exclude_from_get_last_error', False):
     method_call = f'return nidevice_grpc::ApiErrorToStatus(context, status);'
   else:
-    method_call = f'return ConvertApiErrorStatusFor{cpp_handle_type}(context, status, {session});'
+    method_call = f'return {service_deref}ConvertApiErrorStatusFor{cpp_handle_type}(context, status, {session});'
 %>\
       %for error_parameter in error_parameters:
       context->AddTrailingMetadata("${error_parameter["return_on_error_key"]}", std::to_string(${common_helpers.pascal_to_snake(error_parameter["name"])}));

source/custom/nifake_non_ivi_service.custom.cpp

@@ -23,32 +23,50 @@ ::grpc::ServerWriteReactor<ReadStreamResponse>* NiFakeNonIviService::ReadStream(
 
   class ReadStreamReactor : public nidevice_grpc::ServerWriterReactor<ReadStreamResponse, AutoJoinThread> {
    public:
-    ReadStreamReactor(int32 start, int32 stop)
-        : start_(start),
-          stop_(stop)
+    const int32_t MIN_VALUE = 0;
+    const int32_t MAX_VALUE = 1000000;
+
+    ReadStreamReactor(const ReadStreamRequest* request)
     {
-      this->set_producer(
-          start_thread());
+      auto status = start(request);
+      if (!status.ok()) {
+        this->try_finish(std::move(status));
+      }
+    }
+
+    ::grpc::Status start(const ReadStreamRequest* request) {
+      if (request->start() < MIN_VALUE || request->start() > MAX_VALUE) {
+        return ::grpc::Status(::grpc::INVALID_ARGUMENT, "The value for start is out of range.");
+      }
+      if (request->stop() < MIN_VALUE || request->stop() > MAX_VALUE) {
+        return ::grpc::Status(::grpc::INVALID_ARGUMENT, "The value for stop is out of range.");
+      }
+      if (request->start() > request->stop()) {
+        // This error has a different status code and an arbitrary delay for testing purposes.
+        std::this_thread::sleep_for(std::chrono::milliseconds(100));
+        return ::grpc::Status(::grpc::UNKNOWN, "The values for start and stop are reversed.");
+      }
+
+      // Send initial metadata to indicate that the stream has started successfully.
+      StartSendInitialMetadata();
+
+      this->set_producer(start_thread(request->start(), request->stop()));
+      return ::grpc::Status::OK;
     }
 
-    std::unique_ptr<AutoJoinThread> start_thread()
+    std::unique_ptr<AutoJoinThread> start_thread(int32_t start, int32_t stop)
     {
       return AutoJoinThread::make_auto_join(std::thread([=]() {
         ReadStreamResponse response;
-        for (auto i = start_; i <= stop_; ++i) {
+        for (auto i = start; i <= stop; ++i) {
           response.set_value(i);
           queue_write(response);
         }
       }));
     }
-
-    int32 start_;
-    int32 stop_;
   };
 
-  return new ReadStreamReactor(
-      request->start(),
-      request->stop());
+  return new ReadStreamReactor(request);
 }
 
 ::grpc::Status NiFakeNonIviService::ConvertApiErrorStatusForFakeHandle(::grpc::ServerContextBase* context, int32_t status, FakeHandle handle)

source/server/server_reactor.h

@@ -38,9 +38,7 @@ class ServerWriterReactor : public grpc::ServerWriteReactor<TResponse> {
 
   void OnCancel() override
   {
-    LockGuard lock(mutex_);
-    done_ = true;
-    this->Finish(::grpc::Status::OK);
+    try_finish(::grpc::Status::OK);
   }
 
  protected:
@@ -56,6 +54,19 @@ class ServerWriterReactor : public grpc::ServerWriteReactor<TResponse> {
     producer_ = std::move(producer);
   }
 
+  // Try to immediately finish the RPC. If called multiple times, the first call wins.
+  // This method does not wait until the pending send queue is empty.
+  bool try_finish(::grpc::Status status)
+  {
+    LockGuard lock(mutex_);
+    if (!done_) {
+      done_ = true;
+      this->Finish(std::move(status));
+      return true;
+    }
+    return false;
+  }
+
  private:
   void next_write()
   {