C++ Client: End of Barrage stream should throw (rather than segfault)…

…. Also reformat exception messages

cpp-client/deephaven/dhclient/src/subscription/subscribe_thread.cc

@@ -214,7 +214,7 @@ void UpdateProcessor::RunUntilCancelled(std::shared_ptr<UpdateProcessor> self) {
   try {
     self->RunForeverHelper();
   } catch (...) {
-    // If the thread was been cancelled via explicit user action, then swallow all errors.
+    // If the thread has been cancelled via explicit user action, then swallow all errors.
     if (!self->cancelled_) {
       self->callback_->OnFailure(std::current_exception());
     }
@@ -228,6 +228,11 @@ void UpdateProcessor::RunForeverHelper() {
   while (true) {
     auto chunk = fsr_->Next();
     OkOrThrow(DEEPHAVEN_LOCATION_EXPR(chunk));
+    if (chunk->data == nullptr) {
+      // Stream ended. This is abnormal for Deephaven.
+      const char *message = "Unexpected end of stream";
+      throw std::runtime_error(DEEPHAVEN_LOCATION_STR(message));
+    }
     const auto &cols = chunk->data->columns();
     auto column_sources = MakeReservedVector<std::shared_ptr<ColumnSource>>(cols.size());
     auto sizes = MakeReservedVector<size_t>(cols.size());

cpp-client/deephaven/dhcore/src/utility/utility.cc

@@ -104,7 +104,7 @@ void TrueOrThrowHelper(const DebugInfo &debug_info) {
 
 std::string FormatDebugString(const char *func, const char *file, size_t line,
     const std::string &message) {
-  return fmt::format("{}@{}:{}: {}", func, file, line, message);
+  return fmt::format("{}: {}@{}:{}", message, func, file, line);
 }
 
 std::string GetWhat(std::exception_ptr eptr) {