move analysis thread into monitor. populate analysis_it and decoder_it in Prime instead of constantly checking for them. Handle cases where LockedPacket are null due to shutdown

src/zm_monitor.cpp

@@ -395,6 +395,7 @@ Monitor::Monitor()
   storage(nullptr),
   videoStore(nullptr),
   analysis_it(nullptr),
+  analysis_thread(nullptr),
   decoder_it(nullptr),
   decoder(nullptr),
   n_zones(0),
@@ -1126,10 +1127,9 @@ Monitor::~Monitor() {
     disconnect();
   }  // end if mem_ptr
 
-  if (analysis_it) {
+  // Will be free by packetqueue destructor
-    packetqueue.free_it(analysis_it);
   analysis_it = nullptr;
-  }
+  decoder_it = nullptr;
 
   for (int i = 0; i < n_zones; i++) {
     delete zones[i];
@@ -1801,7 +1801,6 @@ bool Monitor::Analyse() {
     Warning("Shouldn't be doing Analyse when not Enabled");
     return false;
   }
-  if (!analysis_it) analysis_it = packetqueue.get_video_it(true);
 
   // if have event, send frames until we find a video packet, at which point do analysis. Adaptive skip should only affect which frames we do analysis on.
 
@@ -1959,12 +1958,12 @@ bool Monitor::Analyse() {
 
               Debug(3, "After motion detection, score:%d last_motion_score(%d), new motion score(%d)",
                   score, last_motion_score, motion_score);
+              motion_frame_count += 1;
             } else {
-              Warning("No image in snap");
+              Debug(1, "No image in snap, codec likely not ready");
             }
             // Why are we updating the last_motion_score too?
             last_motion_score = motion_score;
-            motion_frame_count += 1;
           } else {
             Debug(1, "Skipped motion detection");
           }
@@ -2031,6 +2030,7 @@ bool Monitor::Analyse() {
                 }
                 ZMLockedPacket *lp = packetqueue.get_packet(start_it);
                 delete starting_packet_lock;
+                if (!lp) return false;
                 starting_packet_lock = lp;
                 starting_packet = lp->packet_;
               }
@@ -2129,6 +2129,11 @@ bool Monitor::Analyse() {
                   }
                   ZMLockedPacket *lp = packetqueue.get_packet(start_it);
                   delete starting_packet_lock;
+                  if (!lp) {
+                    // Shutting down event will be closed by ~Monitor()
+                    // Perhaps we shouldn't do this.
+                    return false;
+                  }
                   starting_packet_lock = lp;
                   starting_packet = lp->packet_;
                 }
@@ -2235,6 +2240,7 @@ bool Monitor::Analyse() {
               } // end if zone is alarmed
             } // end foreach zone
           }
+          if (event) {
             if (noteSetMap.size() > 0)
               event->updateNotes(noteSetMap);
             if ( section_length
@@ -2254,6 +2260,9 @@ bool Monitor::Analyse() {
               snprintf(video_store_data->event_file, sizeof(video_store_data->event_file), "%s", event->getEventFile());
               video_store_data->recording = event->StartTime();
             }
+          } else {
+            Error("ALARM but no event");
+          }
 
         } else if ( state == ALERT ) {
           // Alert means this frame has no motion, but we were alarmed and are still recording.
@@ -2735,7 +2744,6 @@ int Monitor::Capture() {
 } // end Monitor::Capture
 
 bool Monitor::Decode() {
-  if (!decoder_it) decoder_it = packetqueue.get_video_it(true);
   
   ZMLockedPacket *packet_lock = packetqueue.get_packet(decoder_it);
   if (!packet_lock) return false;
@@ -3146,8 +3154,17 @@ int Monitor::PrimeCapture() {
       }
     }  // end if rtsp_server
     if (decoding_enabled) {
-      decoder = new DecoderThread(this);
+      if (!decoder_it) decoder_it = packetqueue.get_video_it(false);
+      if (!decoder) decoder = new DecoderThread(this);
     }
+    if (function != MONITOR) {
+      if (!analysis_it) analysis_it = packetqueue.get_video_it(false);
+      if (!analysis_thread) {
+        Debug(1, "Starting an analysis thread for monitor (%d)", id);
+        analysis_thread = new AnalysisThread(this);
+      }
+    }
+
   } else {
     Debug(2, "Failed to prime %d", ret);
   }
@@ -3157,17 +3174,22 @@ int Monitor::PrimeCapture() {
 int Monitor::PreCapture() const { return camera->PreCapture(); }
 int Monitor::PostCapture() const { return camera->PostCapture(); }
 int Monitor::Close() {
+  // Because the stream indexes may change we have to clear out the packetqueue
+  if (decoder) decoder->Stop();
+  if (analysis_thread) analysis_thread->Stop();
+  packetqueue.clear();
   if (decoder) {
-    decoder->Stop();
     delete decoder;
   }
+  if (analysis_thread) {
+    delete analysis_thread;
+  }
   std::lock_guard<std::mutex> lck(event_mutex);
   if (event) {
     Info("%s: image_count:%d - Closing event %" PRIu64 ", shutting down", name, image_count, event->Id());
     closeEvent();
   }
   if (camera) camera->Close();
-  packetqueue.clear();
   return 1;
 }