Heartbeat request support

README.md

@@ -132,6 +132,50 @@ The sampler will automatically add an `X-MEDIA-SEGMENT-DURATION` HTTP response h
 
 In the case of MPEG DASH, the View Results Tree Listener displays the resultant samples with the associated type (manifest, inits and segments for media, audio and subtitles) to easily identify them as well.
 
+## Running periodic requests during playback (Streaming Parallel Controller)
+
+### Concept
+
+Many streaming clients fire a periodic "heartbeat" (analytics beacon, session keep-alive, DRM/license ping, etc.) while the video keeps playing. The **bzm - Streaming Parallel Controller** reproduces that behavior without the CPU and memory cost of the JMeter Parallel Controller: instead of spawning extra threads and deep-cloning the sub-tree, it runs the playback and the heartbeat on the **same thread** by time slicing the playback. Between heartbeats the Streaming Sampler plays in short, resumable slices; every configured interval the controller pauses playback at a slice boundary, runs the heartbeat branch once, and then resumes playback exactly where it left off (no re-download of the master/media playlists or manifest).
+
+![](docs/streaming-parallel-controller.png)
+
+### To add it to your test
+
+- Add the controller: Add -> Logic Controller -> bzm - Streaming Parallel Controller
+- Add a **bzm - Streaming Sampler** as a **direct child** of the controller. This is the playback track.
+- Add the periodic requests (the heartbeat) as the remaining children. They can be plain HTTP Samplers or grouped under a Transaction/Logic Controller.
+
+The test tree looks like this:
+
+![](docs/streaming-parallel-controller-tree.png)
+
+### Controller configuration
+
+#### Heartbeat interval
+
+Set how often (in seconds) the heartbeat branch runs while playback continues in between. For example, `10` fires the heartbeat branch once every 10 seconds of playback.
+
+![](docs/streaming-parallel-controller-interval.png)
+
+#### Run heartbeat immediately
+
+When enabled, the heartbeat branch runs once at the very start of the iteration (before the first interval elapses). When disabled (default), the controller waits one full interval before the first heartbeat.
+
+![](docs/streaming-parallel-controller-run-immediately.png)
+
+### Results
+
+The Streaming Sampler still emits its usual master/media/segment samples, and the heartbeat branch emits its own samples interleaved with them, so in the View Results Tree you see the heartbeat requests appear roughly every interval, in between the segment downloads of a single ongoing playback.
+
+> [!NOTE]
+> **Limitations**
+> - The Streaming Sampler must be a **direct child** of the controller. Do not wrap it in a Transaction Controller, or you would get one tiny transaction per playback slice. If no direct-child Streaming Sampler is found, the controller emits a single failed sample so the misconfiguration is visible.
+> - Heartbeat and playback share **one thread**. A slow heartbeat (large body / slow server) delays segment polling, so this is best suited to small beacon-style requests. For very short intervals (5-10s) on slow networks, a heartbeat may slip by up to one segment download, since a segment download in progress is not interrupted mid-request.
+> - Do **not** add Timers to the heartbeat branch: they block the shared thread and add drift.
+> - The heartbeat interval controls when the branch runs **once**; all samplers under it run sequentially each time. To spread requests evenly, structure the branch accordingly.
+> - Nesting one Streaming Parallel Controller inside another is not supported (the inner one logs a warning and runs its children without independent slicing).
+
 ## Memory tuning: response data release
 
 To reduce JVM memory usage during large/long load tests, the sampler can release

src/main/java/com/blazemeter/jmeter/hls/logic/HlsSampler.java

@@ -2,6 +2,7 @@
 
 import com.blazemeter.jmeter.videostreaming.core.Protocol;
 import com.blazemeter.jmeter.videostreaming.core.SampleResultProcessor;
+import com.blazemeter.jmeter.videostreaming.core.StreamingSliceCoordinator;
 import com.blazemeter.jmeter.videostreaming.core.TimeMachine;
 import com.blazemeter.jmeter.videostreaming.core.Variants;
 import com.blazemeter.jmeter.videostreaming.core.VariantsProvider;
@@ -278,15 +279,28 @@ public SampleResult sample() {
     }
 
     String url = getMasterUrl();
+    boolean active = StreamingSliceCoordinator.isActive();
+    // consume the new-session flag once here; the protocol sampler decides setup vs resume purely
+    // from whether its playback session is still initialized
+    boolean newSession = active && StreamingSliceCoordinator.consumeNewSession();
     if (!url.equals(lastMasterUrl)) {
       try {
         sampler = factory
             .getVideoStreamingSampler(url, this, httpClient, timeMachine, sampleResultProcessor);
       } catch (IllegalArgumentException e) {
         LOG.error("Error initializing the sampler", e);
       }
-    } else if (!this.getResumeVideoStatus()) {
-      sampler.resetVideoStatus();
+    } else if (!active) {
+      if (!this.getResumeVideoStatus()) {
+        sampler.resetVideoStatus();
+      }
+    } else if (newSession) {
+      // new controller iteration on the same URL: honor the resume checkbox across iterations and
+      // drop any cached playlists/manifest so the session starts fresh
+      if (!this.getResumeVideoStatus()) {
+        sampler.resetVideoStatus();
+      }
+      sampler.clearPlaybackSession();
     }
     lastMasterUrl = url;
     return sampler.sample();
@@ -317,6 +331,7 @@ public boolean interrupt() {
 
   @Override
   public void threadFinished() {
+    StreamingSliceCoordinator.clear();
     httpClient.threadFinished();
   }
 

src/main/java/com/blazemeter/jmeter/videostreaming/core/PlaybackSession.java

@@ -0,0 +1,66 @@
+package com.blazemeter.jmeter.videostreaming.core;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Single owner of the playback state that must persist across slices within one controller
+ * session, so a continuing session does not re-download the master/variant playlists (or manifest)
+ * or rebuild the per-track playbacks on every slice.
+ *
+ * <p>Parameterized over the protocol-specific playback type {@code P} and top-level document type
+ * {@code M} (HLS master playlist / DASH manifest) so a single class serves both protocols while the
+ * owning sampler keeps full type safety.
+ *
+ * @param <P> protocol-specific media playback type
+ * @param <M> protocol-specific top-level document type (manifest / master playlist)
+ */
+public class PlaybackSession<P, M> {
+
+  private boolean initialized;
+  private P primary;
+  private final List<P> complements = new ArrayList<>();
+  private M manifest;
+
+  public boolean isInitialized() {
+    return initialized;
+  }
+
+  public void setPrimary(P primary) {
+    this.primary = primary;
+  }
+
+  public P getPrimary() {
+    return primary;
+  }
+
+  public void setComplements(List<P> playbacks) {
+    complements.clear();
+    complements.addAll(playbacks);
+  }
+
+  public List<P> getComplements() {
+    return complements;
+  }
+
+  public void setManifest(M manifest) {
+    this.manifest = manifest;
+  }
+
+  public M getManifest() {
+    return manifest;
+  }
+
+  /** Marks the session as usable for resume. Call only after setup fully succeeds. */
+  public void markInitialized() {
+    this.initialized = true;
+  }
+
+  /** Releases all retained playback state. Call on real end, error, interrupt or setup failure. */
+  public void clear() {
+    initialized = false;
+    primary = null;
+    complements.clear();
+    manifest = null;
+  }
+}

src/main/java/com/blazemeter/jmeter/videostreaming/core/StreamingSliceCoordinator.java

@@ -0,0 +1,99 @@
+package com.blazemeter.jmeter.videostreaming.core;
+
+/**
+ * Per-thread handoff between the Streaming Parallel Controller and the streaming sampler.
+ *
+ * <p>The controller owns the lifecycle ({@link #beginIteration()}) and
+ * the slice deadline; the sampler only reads the deadline (via {@link #shouldYield()} /
+ * {@link #clampMillis(long)}) and writes a single {@link SliceExit} in a {@code finally} block.
+ *
+ * <p>When no controller is driving the current thread the coordinator behaves as a null object:
+ * {@link #isActive()} and {@link #shouldYield()} return {@code false} and
+ * {@link #clampMillis(long)} returns its argument unchanged, so standalone behavior is untouched.
+ */
+public final class StreamingSliceCoordinator {
+
+  public enum SliceExit {
+    YIELD, FINISHED, ERROR, INTERRUPTED
+  }
+
+  private static final class State {
+
+    private boolean active;
+    private boolean newSession;
+    private long sliceDeadlineNanos = Long.MAX_VALUE;
+    private SliceExit lastExit;
+  }
+
+  private static final ThreadLocal<State> STATE = ThreadLocal.withInitial(State::new);
+
+  /**
+   * Called by the controller at the start of each of its iterations. Marks the thread as sliced and
+   * signals a new playback session.
+   */
+  public static void beginIteration() {
+    State s = STATE.get();
+    s.active = true;
+    s.newSession = true;
+    s.sliceDeadlineNanos = Long.MAX_VALUE;
+    s.lastExit = null;
+  }
+
+  public static void setDeadlineNanos(long deadlineNanos) {
+    STATE.get().sliceDeadlineNanos = deadlineNanos;
+  }
+
+  public static boolean isActive() {
+    return STATE.get().active;
+  }
+
+  /**
+   * Reads and clears the new-session flag. Only meaningful while {@link #isActive()}.
+   */
+  public static boolean consumeNewSession() {
+    State s = STATE.get();
+    boolean value = s.newSession;
+    s.newSession = false;
+    return value;
+  }
+
+  /**
+   * True when a controller is slicing and the current slice deadline has passed.
+   */
+  public static boolean shouldYield() {
+    State s = STATE.get();
+    return s.active && System.nanoTime() >= s.sliceDeadlineNanos;
+  }
+
+  /**
+   * Bounds a requested await so the streaming loop never sleeps past the current slice deadline.
+   * Returns the request unchanged when no controller is active.
+   */
+  public static long clampMillis(long requestedMillis) {
+    State s = STATE.get();
+    if (!s.active) {
+      return requestedMillis;
+    }
+    long remaining = (s.sliceDeadlineNanos - System.nanoTime()) / 1_000_000L;
+    if (remaining <= 0) {
+      return 0;
+    }
+    return Math.min(requestedMillis, remaining);
+  }
+
+  public static void setExit(SliceExit exit) {
+    STATE.get().lastExit = exit;
+  }
+
+  public static SliceExit getExit() {
+    return STATE.get().lastExit;
+  }
+
+  public static void clear() {
+    State s = STATE.get();
+    s.active = false;
+    s.newSession = false;
+    s.sliceDeadlineNanos = Long.MAX_VALUE;
+    s.lastExit = null;
+  }
+}