feature: [158] Implement global mqtt message storage service, part 2

[JavaSaBr]

Issue

#15

Description

This branch extends the publish pipeline around global MQTT message storage and refactors the broker to distinguish more clearly between incoming publishes, outgoing publishes, and the storage/lifecycle around them.

Main changes

1. Introduced dedicated incoming publish storage

• Added IncomingPublishStorage and InMemoryIncomingPublishStorage.
• Incoming publishes now have explicit lifecycle operations:
  • store(...)
  • remove(...)
  • removeIfExist(...)
  • increaseConsumerCount(...)
  • decreaseConsumerCount(...)
• The storage now tracks registered incoming publishes separately from payload storage.

2. Split publish models into incoming and outgoing variants

• Added:
  • SimpleIncomingPublish
  • SimpleOutgoingPublish
  • TrackableSimpleOutgoingPublish
• Updated the model layer so IncomingPublish and OutgoingPublish are distinct concepts built on the shared Publish contract.
• This makes the delivery pipeline more explicit:
  • an incoming publish represents a broker-received message
  • an outgoing publish represents a subscriber-delivery form of that message

3. Reworked publish handling flow

• PublishMqttInMessageHandler now:
  • stores PublishData first
  • stores the IncomingPublish
  • removes stored PublishData if incoming publish registration fails
• The handler now uses session.generatePublishId() to assign a unique publish identity for storage and tracking.

4. Updated incoming publish processors

• AbstractIncomingPublishProcessor now works with IncomingPublish directly and dispatches through storage-aware lifecycle rules.
• Subscriber fan-out now uses consumer counting in IncomingPublishStorage.
• When no subscribers match, the processor removes the publish via storage-aware logic.
• The QoS processors were updated to follow the new incoming-publish model.
• The QoS 2 flow was reworked so dispatching begins only after the publish is confirmed/received in the intended phase.

5. Extended session state for stored publish tracking

• MqttSession now exposes generatePublishId().
• InMemoryNetworkMqttSession now maintains a publish-id generator in addition to message-id and data-id generators.
• Session-side processing structures were updated to support the new incoming/outgoing publish lifecycle.

6. Updated publisher-to-subscriber delivery model

• Subscriber publish senders now work with the new outgoing publish types rather than reusing the incoming publish object shape directly.
• Trackable outgoing publishes are now represented explicitly for QoS flows that require retry/ack tracking.

Documentation and repo conventions

• .github/copilot-instructions.md was expanded with testing and stateful-service conventions observed in this branch:
  • explicit test fixture naming
  • using domain constants instead of magic values
  • using shared integration fixtures intentionally
  • creating stateful test objects through real storage/service paths
  • preferring strict managed-state contracts for stateful services

Overall impact

This branch lays the groundwork for a broker-wide message-storage model by:

1. separating incoming and outgoing publish representations,
2. storing incoming publishes explicitly,
3. wiring publish storage into message handling and processor flow,
4. introducing publish identity generation at the session level,
5. and updating integration tests to follow the new managed-state lifecycle.

[github-actions]

Test Coverage Report

Overall Project	85.02% -0.91%	🍏
Files changed	79.94%	🍏

File	Coverage
ConcurrentRetainedMessageTree.java	100%	🍏
TrackableIncomingPublishProcessor.java	100%	🍏
Qos0SubscriberPublishSender.java	100%	🍏
SubscribeMqttInMessageHandler.java	99.77%	🍏
RetainedPublishNode.java	97.66%	🍏
SimpleIncomingPublish.java	96.12% -3.88%	🍏
InMemoryNetworkMqttSession.java	95.35%	🍏
MqttBrokerSpringConfig.java	88.8%	🍏
DefaultPublishDispatcher.java	84.38%	🍏
DefaultIncomingPublishRouter.java	83.68%	🍏
Qos1IncomingPublishProcessor.java	82.78%	🍏
PublishMqttInMessageHandler.java	81.63% -2.72%	🍏
InMemoryRetainPublishService.java	80%	🍏
InMemoryProcessingPublishes.java	78.76%	🍏
Qos1SubscriberPublishSender.java	78.61%	🍏
InMemoryIncomingPublishStorage.java	77.9% -22.1%	🍏
Qos2SubscriberPublishSender.java	75.95%	🍏
TrackableSubscriberPublishSender.java	73.19%	🍏
AbstractIncomingPublishProcessor.java	70.65% -5.8%	🍏
TrackableSimpleOutgoingPublish.java	68.67% -31.33%	🍏
SimpleOutgoingPublish.java	62.07% -37.93%	🍏
AbstractSubscriberPublishSender.java	60.75%	🍏
Qos2IncomingPublishProcessor.java	60.72% -10.85%	🍏
Qos0IncomingPublishProcessor.java	57.89% -14.04%	❌
InMemoryPublishDataStorage.java	48.15% -14.81%	🍏

[crazyrokr]

Could we consider extracting of common part from all children into Publish interface. Something like this:

interface Publish {
  public Publish withoutRetained() {
      if (!retained()) {
        return this;
      } else {
        return createPublishWithoutRetained();
  }
}

and

record TrackableSimpleOutgoingPublish() {
  public Publish createPublishWithoutRetained(){
    return new TrackableSimpleOutgoingPublish(incomingPublish, messageId, duplicated, false, qos, subscriptionIds)
  }
}

[JavaSaBr]

I don't see any profit from this + we talk about records without inheritance, so your way doesn't solve anything

[crazyrokr]

Does it worth to do in a separate read lock ?

[JavaSaBr]

@crazyrokr we don't expect any collisions, so using a read lock here is overhead

[crazyrokr]

The same as previous

[JavaSaBr]

@crazyrokr the same answer

[crazyrokr]

The same as previous

[crazyrokr]

Should it be IncomingPublish?

[JavaSaBr]

@crazyrokr currently the lambda doesn't support this type