Add log cache log streaming support

README.md

@@ -17,6 +17,41 @@ The `cf-java-client` project is a Java language binding for interacting with a C
 ## Versions
 The Cloud Foundry Java Client has two active versions. The `5.x` line is compatible with Spring Boot `2.4.x - 2.6.x` just to manage its dependencies, while the `4.x` line uses Spring Boot `2.3.x`.
 
+## Deprecations
+
+### `DopplerClient.recentLogs()` — Recent Logs via Doppler
+
+> [!WARNING]
+> **Deprecated since cf-java-client `5.17.x`**
+>
+> The `DopplerClient.recentLogs()` endpoint (and the related `RecentLogsRequest` / `LogMessage` types from the `org.cloudfoundry.doppler` package) are **deprecated** and will be removed in a future release.
+>
+> This API relies on the [Loggregator][loggregator] Doppler/Traffic Controller endpoint `/apps/{id}/recentlogs`, which was removed in **Loggregator ≥ 107.0**.
+> The affected platform versions are:
+>
+> | Platform | Last version with Doppler recent-logs support |
+> | -------- | --------------------------------------------- |
+> | CF Deployment (CFD) | `< 24.3` |
+> | Tanzu Application Service (TAS) | `< 4.0` |
+>
+> **Migration:** Replace any call to `DopplerClient.recentLogs()` with [`LogCacheClient.read()`][log-cache-api] (available via `org.cloudfoundry.logcache.v1.LogCacheClient`).
+>
+> ```java
+> // Before (deprecated)
+> dopplerClient.recentLogs(RecentLogsRequest.builder()
+>     .applicationId(appId)
+>     .build());
+>
+> // After
+> logCacheClient.read(ReadRequest.builder()
+>     .sourceId(appId)
+>     .envelopeTypes(EnvelopeType.LOG)
+>     .build());
+> ```
+
+[loggregator]: https://github.com/cloudfoundry/loggregator
+[log-cache-api]: https://github.com/cloudfoundry/log-cache
+
 ## Dependencies
 Most projects will need two dependencies; the Operations API and an implementation of the Client API.  For Maven, the dependencies would be defined like this:
 
@@ -76,6 +111,9 @@ Both the `cloudfoundry-operations` and `cloudfoundry-client` projects follow a [
 
 ### `CloudFoundryClient`, `DopplerClient`, `UaaClient` Builders
 
+> [!NOTE]
+> **`DopplerClient` — partial deprecation:** The `recentLogs()` method on `DopplerClient` is deprecated and only works against Loggregator \< 107.0 (CFD \< 24.3 / TAS \< 4.0). See the [Deprecations](#deprecations) section above for the migration path to `LogCacheClient`.
+
 The lowest-level building blocks of the API are `ConnectionContext` and `TokenProvider`.  These types are intended to be shared between instances of the clients, and come with out of the box implementations.  To instantiate them, you configure them with builders:
 
 ```java
@@ -297,6 +335,7 @@ Name | Description
 `TEST_PROXY_PORT` | _(Optional)_ The port of a proxy to route all requests through. Defaults to `8080`.
 `TEST_PROXY_USERNAME` | _(Optional)_ The username for a proxy to route all requests through
 `TEST_SKIPSSLVALIDATION` | _(Optional)_ Whether to skip SSL validation when connecting to the Cloud Foundry instance.  Defaults to `false`.
+`UAA_API_REQUEST_LIMIT` | _(Optional)_ If your UAA server does rate limiting and returns 429 errors, set this variable to the smallest limit configured there. Whether your server limits UAA calls is shown in the log, together with the location of the configuration file on the server. Defaults to `0` (no limit).
 
 If you do not have access to a CloudFoundry instance with admin access, you can run one locally using [bosh-deployment](https://github.com/cloudfoundry/bosh-deployment) & [cf-deployment](https://github.com/cloudfoundry/cf-deployment/) and Virtualbox.
 

cloudfoundry-client-reactor/pom.xml

@@ -25,7 +25,7 @@
     <parent>
         <groupId>org.cloudfoundry</groupId>
         <artifactId>cloudfoundry-java-client</artifactId>
-        <version>5.15.0.BUILD-SNAPSHOT</version>
+        <version>6.0.0-SNAPSHOT</version>
     </parent>
 
     <artifactId>cloudfoundry-client-reactor</artifactId>

cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/client/_ReactorCloudFoundryClient.java

@@ -76,6 +76,7 @@
 import org.cloudfoundry.client.v3.spaces.SpacesV3;
 import org.cloudfoundry.client.v3.stacks.StacksV3;
 import org.cloudfoundry.client.v3.tasks.Tasks;
+import org.cloudfoundry.client.v3.users.UsersV3;
 import org.cloudfoundry.reactor.ConnectionContext;
 import org.cloudfoundry.reactor.TokenProvider;
 import org.cloudfoundry.reactor.client.v2.applications.ReactorApplicationsV2;
@@ -136,6 +137,7 @@
 import org.cloudfoundry.reactor.client.v3.spaces.ReactorSpacesV3;
 import org.cloudfoundry.reactor.client.v3.stacks.ReactorStacksV3;
 import org.cloudfoundry.reactor.client.v3.tasks.ReactorTasks;
+import org.cloudfoundry.reactor.client.v3.users.ReactorUsersV3;
 import org.immutables.value.Value;
 import reactor.core.publisher.Mono;
 
@@ -509,6 +511,12 @@ public Users users() {
         return new ReactorUsers(getConnectionContext(), getRootV2(), getTokenProvider(), getRequestTags());
     }
 
+    @Override
+    @Value.Derived
+    public UsersV3 usersV3() {
+        return new ReactorUsersV3(getConnectionContext(), getRootV3(), getTokenProvider(), getRequestTags());
+    }
+
     /**
      * The connection context
      */

cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/client/v3/quotas/organizations/ReactorOrganizationQuotasV3.java

@@ -94,4 +94,18 @@ public Mono<String> delete(DeleteOrganizationQuotaRequest request) {
                                         "organization_quotas", request.getOrganizationQuotaId()))
                 .checkpoint();
     }
+
+    @Override
+    public Mono<ApplyOrganizationQuotaResponse> apply(ApplyOrganizationQuotaRequest request) {
+        return post(
+                        request,
+                        ApplyOrganizationQuotaResponse.class,
+                        builder ->
+                                builder.pathSegment(
+                                        "organization_quotas",
+                                        request.getOrganizationQuotaId(),
+                                        "relationships",
+                                        "organizations"))
+                .checkpoint();
+    }
 }

cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/client/v3/quotas/spaces/ReactorSpaceQuotasV3.java

@@ -87,4 +87,33 @@ public Mono<String> delete(DeleteSpaceQuotaRequest request) {
                         builder -> builder.pathSegment("space_quotas", request.getSpaceQuotaId()))
                 .checkpoint();
     }
+
+    @Override
+    public Mono<ApplySpaceQuotaResponse> apply(ApplySpaceQuotaRequest request) {
+        return post(
+                        request,
+                        ApplySpaceQuotaResponse.class,
+                        builder ->
+                                builder.pathSegment(
+                                        "space_quotas",
+                                        request.getSpaceQuotaId(),
+                                        "relationships",
+                                        "spaces"))
+                .checkpoint();
+    }
+
+    @Override
+    public Mono<Void> remove(RemoveSpaceQuotaRequest request) {
+        return delete(
+                        request,
+                        Void.class,
+                        builder ->
+                                builder.pathSegment(
+                                        "space_quotas",
+                                        request.getSpaceQuotaId(),
+                                        "relationships",
+                                        "spaces",
+                                        request.getSpaceId()))
+                .checkpoint();
+    }
 }

cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/client/v3/users/ReactorUsersV3.java

@@ -0,0 +1,85 @@
+/*
+ * Copyright 2013-2025 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.cloudfoundry.reactor.client.v3.users;
+
+import java.util.Map;
+import org.cloudfoundry.client.v3.users.*;
+import org.cloudfoundry.reactor.ConnectionContext;
+import org.cloudfoundry.reactor.TokenProvider;
+import org.cloudfoundry.reactor.client.v3.AbstractClientV3Operations;
+import reactor.core.publisher.Mono;
+
+/**
+ * The Reactor-based implementation of {@link UsersV3}
+ */
+public class ReactorUsersV3 extends AbstractClientV3Operations implements UsersV3 {
+
+    /**
+     * Creates an instance
+     *
+     * @param connectionContext the {@link ConnectionContext} to use when communicating with the server
+     * @param root              the root URI of the server. Typically, something like {@code https://api.cloudfoundry.your.company.com}.
+     * @param tokenProvider     the {@link TokenProvider} to use when communicating with the server
+     * @param requestTags       map with custom http headers which will be added to web request
+     */
+    public ReactorUsersV3(
+            ConnectionContext connectionContext,
+            Mono<String> root,
+            TokenProvider tokenProvider,
+            Map<String, String> requestTags) {
+        super(connectionContext, root, tokenProvider, requestTags);
+    }
+
+    @Override
+    public Mono<CreateUserResponse> create(CreateUserRequest request) {
+        return post(request, CreateUserResponse.class, builder -> builder.pathSegment("users"))
+                .checkpoint();
+    }
+
+    @Override
+    public Mono<GetUserResponse> get(GetUserRequest request) {
+        return get(
+                        request,
+                        GetUserResponse.class,
+                        builder -> builder.pathSegment("users", request.getUserId()))
+                .checkpoint();
+    }
+
+    @Override
+    public Mono<ListUsersResponse> list(ListUsersRequest request) {
+        return get(request, ListUsersResponse.class, builder -> builder.pathSegment("users"))
+                .checkpoint();
+    }
+
+    @Override
+    public Mono<UpdateUserResponse> update(UpdateUserRequest request) {
+        return patch(
+                        request,
+                        UpdateUserResponse.class,
+                        builder -> builder.pathSegment("users", request.getUserId()))
+                .checkpoint();
+    }
+
+    @Override
+    public Mono<Void> delete(DeleteUserRequest request) {
+        return delete(
+                        request,
+                        Void.class,
+                        builder -> builder.pathSegment("users", request.getUserId()))
+                .checkpoint();
+    }
+}

cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/logcache/v1/ReactorLogCacheEndpoints.java

@@ -16,15 +16,23 @@
 
 package org.cloudfoundry.reactor.logcache.v1;
 
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+import org.cloudfoundry.logcache.v1.Envelope;
+import org.cloudfoundry.logcache.v1.EnvelopeType;
 import org.cloudfoundry.logcache.v1.InfoRequest;
 import org.cloudfoundry.logcache.v1.InfoResponse;
 import org.cloudfoundry.logcache.v1.MetaRequest;
 import org.cloudfoundry.logcache.v1.MetaResponse;
 import org.cloudfoundry.logcache.v1.ReadRequest;
 import org.cloudfoundry.logcache.v1.ReadResponse;
+import org.cloudfoundry.logcache.v1.TailLogsRequest;
 import org.cloudfoundry.reactor.ConnectionContext;
 import org.cloudfoundry.reactor.TokenProvider;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 final class ReactorLogCacheEndpoints extends AbstractLogCacheOperations {
@@ -48,4 +56,111 @@ Mono<MetaResponse> meta(MetaRequest request) {
     Mono<ReadResponse> read(ReadRequest request) {
         return get(request, ReadResponse.class, "read", request.getSourceId()).checkpoint();
     }
+
+    Mono<ReadResponse> recentLogs(ReadRequest request) {
+        return read(request);
+    }
+
+    /**
+     * Continuously polls Log Cache and emits new {@link Envelope}s as they arrive.
+     *
+     * <p>Mirrors the Go {@code logcache.Walk()} / {@code cf tail --follow} semantics:
+     * <ol>
+     *   <li>Start the cursor at {@code startTime} (defaults to now&nbsp;&minus;&nbsp;5&nbsp;s in
+     *       nanoseconds).</li>
+     *   <li>Issue {@code GET /api/v1/read/{sourceId}?start_time=cursor}.</li>
+     *   <li>Emit every returned envelope in ascending timestamp order and advance
+     *       the cursor to {@code lastTimestamp + 1}.</li>
+     *   <li>When the batch is empty, wait {@code pollInterval} before the next poll.</li>
+     *   <li>Repeat forever – the caller cancels the subscription to stop.</li>
+     * </ol>
+     * Fully non-blocking: no {@code Thread.sleep}.
+     */
+    Flux<Envelope> logsTail(TailLogsRequest request) {
+        long defaultStartNanos = (System.currentTimeMillis() - 5_000L) * 1_000_000L;
+        AtomicLong cursor =
+                new AtomicLong(
+                        request.getStartTime() != null
+                                ? request.getStartTime()
+                                : defaultStartNanos);
+
+        List<EnvelopeType> envelopeTypes =
+                request.getEnvelopeTypes() != null
+                        ? request.getEnvelopeTypes()
+                        : Collections.emptyList();
+        String nameFilter = request.getNameFilter();
+
+        /*
+         * Strategy (mirrors Go's logcache.Walk):
+         *  – Mono.defer builds a fresh ReadRequest from the mutable cursor on every repetition.
+         *  – The Mono returns either the sorted batch (non-empty) or an empty list.
+         *  – flatMapMany turns each batch into a stream of individual Envelope items.
+         *  – repeat() subscribes again after each completion.
+         *  – When the batch was empty we insert a delay via Mono.delay before the next
+         *    repetition so we do not hammer the server. We signal "empty" by returning
+         *    a sentinel Mono<Boolean> (false = was empty, true = had data) and use
+         *    repeatWhen to conditionally delay.
+         */
+        return Flux.defer(
+                        () -> {
+                            // Build the read request from the current cursor position.
+                            ReadRequest.Builder builder =
+                                    ReadRequest.builder()
+                                            .sourceId(request.getSourceId())
+                                            .startTime(cursor.get());
+                            if (!envelopeTypes.isEmpty()) {
+                                builder.envelopeTypes(envelopeTypes);
+                            }
+                            if (nameFilter != null && !nameFilter.isEmpty()) {
+                                builder.nameFilter(nameFilter);
+                            }
+
+                            return read(builder.build())
+                                    .onErrorReturn(ReadResponse.builder().build())
+                                    .flatMapMany(
+                                            resp -> {
+                                                List<Envelope> raw =
+                                                        resp.getEnvelopes() != null
+                                                                ? resp.getEnvelopes().getBatch()
+                                                                : Collections.emptyList();
+
+                                                if (raw.isEmpty()) {
+                                                    // Signal "no data" so repeatWhen can insert the
+                                                    // back-off delay.
+                                                    return Flux.empty();
+                                                }
+
+                                                // Sort ascending by timestamp and advance the
+                                                // cursor.
+                                                List<Envelope> sorted = new ArrayList<>(raw);
+                                                sorted.sort(
+                                                        (a, b) ->
+                                                                Long.compare(
+                                                                        a.getTimestamp() != null
+                                                                                ? a.getTimestamp()
+                                                                                : 0L,
+                                                                        b.getTimestamp() != null
+                                                                                ? b.getTimestamp()
+                                                                                : 0L));
+
+                                                Envelope last = sorted.get(sorted.size() - 1);
+                                                cursor.set(
+                                                        (last.getTimestamp() != null
+                                                                        ? last.getTimestamp()
+                                                                        : cursor.get())
+                                                                + 1);
+
+                                                return Flux.fromIterable(sorted);
+                                            });
+                        })
+                // repeatWhen receives a Flux<Long> where each element is the count of items
+                // emitted in the previous cycle (0 = empty batch → insert delay).
+                .repeatWhen(
+                        companion ->
+                                companion.flatMap(
+                                        count ->
+                                                count == 0
+                                                        ? Mono.delay(request.getPollInterval())
+                                                        : Mono.just(count)));
+    }
 }