fix(drain): notify services and clients, and report the batch count

[benaliabderrahmane]

Description

on_new_request_cb and on_new_response_cb were stored by their setters and
flushed once against an already-queued backlog, but no drain ever fired them
again. A service or client was callback-dead the moment its registration
returned: every request and response after that arrived without notification.

Now that one drain fills all three queues, populating the callback fields in the
two drain_target() overloads is all that bug needed. Two corrections to the
notification protocol ride along with it, one in drain_endpoint() and one in
all three set_on_new_*_callback() setters.

The notification also moves out of the receive loop. It fired once per datagram
with a count of 1; it now fires once per drain that grew the queue, carrying
the number of entries it gained. rmw/event_callback_type.h defines
number_of_events as the count since the callback was last called, explicitly
allows > 1, and says it should never be 0.

The count is how much the queue grew — enqueued - dropped — not how many
datagrams the socket handed over. number_of_events is a take credit: the
executor calls take() once per event it is told about, so an overflow that
pushed 100 and popped 90 must report 10, or the executor spends 90 takes finding
nothing. There is at most one pop per push, so the subtraction never underflows,
and a drain whose net gain is zero now stays silent for the same reason an empty
drain does.

The second correction is in the setters: all three flush the queued backlog only
when a callback takes over from none. rclcpp installs the callback twice in a
row on purpose — once against a stack temporary, then again against its
permanent storage, to close the window where the std::function is replaced but
the middleware still holds the old pointer (subscription_base.hpp) — and an
unconditional flush pays the same backlog out for both calls. A TRANSIENT_LOCAL
subscription with ten replayed samples was handing the executor twenty events.

Is this user-facing behavior change?

Yes:

• Service and client listener callbacks now fire on every drain that grows the
  queue, not just on the setter's one-shot flush.
• number_of_events is now a batch count rather than always 1. A drain that
  gains ten entries reports ten once instead of one ten times, and a drain whose
  arrivals were all popped again by the overflow trim reports nothing at all.
• An endpoint that already has messages queued when its callback is installed
  reports that backlog once instead of twice. Because rclcpp registers twice in
  a row, a TRANSIENT_LOCAL subscription with ten replayed samples reported
  twenty events and now reports ten.

One judgment call for review: the flush fires only on the none→callback
transition, so replacing a non-null callback with a different non-null one
while a backlog is queued hands the new callback nothing — it hears about those
entries only on the next drain that grows the queue. That is the cost of closing
the double payout, and the re-registration rclcpp actually performs is the case
the guard was written for, but if we would rather flush on every transition into
a non-null callback, this is the line to say so on.

How was this tested?

New test/test_rmw_listener_callbacks.cpp, 8 tests: all three endpoint types
notified from an rmw_wait-driven drain, the batch count, that the count
excludes datagrams the overflow trim popped, that a second registration does not
re-flush the backlog, the never-zero rule, and that clearing a callback stops
notification.

Five of the eight fail on the parent commit: both service and client tests
see no callback at all; SubscriptionCallbackReportsTheBatchCount sees three
calls of 1 instead of one call of 3;
BatchCountExcludesDatagramsDroppedByOverflow sees 15 events for a depth-10
queue instead of 10; and ReRegisteringTheCallbackDoesNotReflushTheBacklog sees
6 events in 2 calls for 3 queued messages instead of 3 in 1.

Full suite on jazzy: 165 tests, 0 failures. kilted, rolling and lyrical are
on CI.

Did you use Generative AI?

Additional Information

Builds on the drain unification PR; merge that first.

[benaliabderrahmane]

lgtm!