| 277 | } |
| 278 | |
| 279 | private T applyUpdate(Function<T, T> updateFn) { |
| 280 | rwLock.writeLock().lock(); |
| 281 | T newValue; |
| 282 | try { |
| 283 | value = updateFn.apply(value); |
| 284 | newValue = value; |
| 285 | } finally { |
| 286 | rwLock.writeLock().unlock(); |
| 287 | } |
| 288 | SubmissionPublisher<T> p = changesPublisher; |
| 289 | if (p != null) { |
| 290 | // Two distinct edge cases handled here, on purpose: |
| 291 | // |
| 292 | // (a) Slow-subscriber drop policy. The onDrop predicate |
| 293 | // `(sub, item) -> false` tells SubmissionPublisher.offer |
| 294 | // NOT to retry when a subscriber's per-subscriber buffer |
| 295 | // is full — i.e. drop the emission for that lagging |
| 296 | // subscriber. This is the documented contract: the agent's |
| 297 | // serial update loop must never back-pressure on a slow |
| 298 | // downstream. |
| 299 | // |
| 300 | // (b) Close-vs-offer race. Between this thread reading |
| 301 | // `changesPublisher` (above) and calling `offer`, another |
| 302 | // thread can run `shutdown()` and close the publisher. The |
| 303 | // resulting IllegalStateException is intentionally |
| 304 | // swallowed: the publisher is gone, this emission is |
| 305 | // moot, and the close path is responsible for |
| 306 | // onComplete-ing remaining subscribers. |
| 307 | try { |
| 308 | p.offer(newValue, (sub, item) -> false); |
| 309 | } catch (IllegalStateException ignore) { |
| 310 | // see (b) above |
| 311 | } |
| 312 | } |
| 313 | return newValue; |
| 314 | } |
| 315 | |
| 316 | /** |
| 317 | * An executor that serializes task execution on a delegate executor. |