Add log cache log streaming support

[ZPascal]

Summary

Adds live log streaming via Log Cache — the Java equivalent of cf tail --follow.

Previously the only Log Cache read path was logsRecent, a single snapshot request. There was no way to continuously receive new log envelopes without polling manually. This change introduces a logsTail API across all three relevant modules.

---

Motivation

Cloud Foundry dropped the legacy Loggregator Doppler streaming endpoint (DopplerClient.stream()) in Loggregator ≥ 107.0 (CFD ≥ 24.3 / TAS ≥ 4.0). The CF CLI replaced it with a polling loop over the Log Cache /api/v1/read endpoint — that is what cf tail --follow does today. This PR brings the same capability to Java consumers.

---

Changes

Changes in detail

cloudfoundry-client — new TailLogsRequest and LogCacheClient.logsTail()

New _TailLogsRequest.java (Immutables @Value.Immutable):

Field	Type	Default	Description
sourceId	String	—	App / service GUID (required)
startTime	Long (nullable)	now − 5 s (ns)	Cursor start time in UNIX nanoseconds
envelopeTypes	List<EnvelopeType> (nullable)	all types	Envelope type filter
nameFilter	String (nullable)	none	Regex name filter (Log Cache ≥ 2.1.0)
pollInterval	Duration	250 ms	Back-off between polls when no new data

New method on LogCacheClient:

/**
 * Continuously polls Log Cache /api/v1/read and streams new Envelopes as they appear.
 * Equivalent to the Go logcache.Walk() API and `cf tail --follow`.
 * The Flux never completes on its own – cancel the subscription to stop streaming.
 */
Flux<Envelope> logsTail(TailLogsRequest request);

---

cloudfoundry-client-reactor — non-blocking polling implementation

ReactorLogCacheEndpoints.logsTail() mirrors the Go logcache.Walk() algorithm, fully non-blocking:

1. An AtomicLong cursor starts at startTime (or now − 5 s).
2. Flux.defer builds a fresh ReadRequest from the current cursor on every repetition and calls GET /api/v1/read/{sourceId}?start_time=cursor.
3. Envelopes are sorted ascending by timestamp; the cursor advances to lastTimestamp + 1; each envelope is emitted individually downstream.
4. When the batch is empty, repeatWhen inserts a Mono.delay(pollInterval) before the next poll. When envelopes arrive, the next poll starts immediately.
5. The Flux is infinite; the caller cancels the subscription to stop.

Transient network errors are logged and retried. HTTP errors (4xx/5xx) are propagated to the subscriber.

---

cloudfoundry-operations — Applications.logsTail()

DefaultApplications delegates to LogCacheClient:

@Override
public Flux<Envelope> logsTail(TailLogsRequest request) {
    return this.logCacheClient
            .flatMapMany(client -> client.logsTail(request))
            .transform(OperationsLogging.log("Tail Application Logs"))
            .checkpoint();
}

---

Tests

Test	What it verifies
logsTailLogCache	Happy path: a single LOG/OUT envelope is forwarded correctly
logsTailLogCacheMultipleEnvelopes	3 envelopes with types OUT → ERR → OUT are emitted in order
logsTailLogCacheError	A RuntimeException from the client propagates unchanged to the subscriber
logsTailLogCacheOutAndErrEnvelopes	Both stdout and stderr envelopes are forwarded without filtering

---

Usage

TailLogsRequest request = TailLogsRequest.builder()
        .sourceId(applicationGuid)
        .envelopeTypes(List.of(EnvelopeType.LOG))
        .pollInterval(Duration.ofMillis(250))
        .build();

logCacheClient.logsTail(request)
        .filter(e -> e.getLog() != null)
        .map(e -> e.getLog().getPayloadAsText())
        .subscribe(System.out::println);

Or via the Operations API:

cloudFoundryOperations.applications()
        .logsTail(TailLogsRequest.builder().sourceId(appGuid).build())
        .filter(e -> e.getLog() != null)
        .map(e -> e.getLog().getPayloadAsText())
        .subscribe(System.out::println);

---

Relation to existing API

Method	Transport	Completes?	Use case
DopplerClient.stream() ⚠️ deprecated	WebSocket / Doppler	Yes (server closes)	Legacy streaming (Loggregator < 107.0)
LogCacheClient.recentLogs()	HTTP GET (single)	Yes	Fetch last N log lines
LogCacheClient.logsTail() ✅ new	HTTP GET (polling loop)	Never	Live streaming (Loggregator ≥ 107.0)

---

Checklist

• _TailLogsRequest Immutables value object
• LogCacheClient.logsTail() interface method
• ReactorLogCacheEndpoints.logsTail() — non-blocking Reactor implementation
• _ReactorLogCacheClient.logsTail() — delegate override
• Applications.logsTail() — Operations API method
• DefaultApplications.logsTail() — implementation
• Unit tests in DefaultApplicationsTest (120 passing)
• Integration tests