To support exactly once subscribe, we will need to:

• Refactor message sender, lease state to use LeaseInfo
• Introduce AckError (doesn't have to be perfect)
• Add exactly once acking to the Leaser and DefaultLeaser
• Add exactly_once::Leases
  • Accepting acks, nacks
  • Flushing
  • Lease extensions
  • Shutdown
  • Confirm results
• Add confirmed ack to the LeaseState (no retries)
• Add confirmed ack to the ack senders, lease loop
• Add exactly once handler with ack(), nack()
• Add confirmed_ack() to the handler
• Add ExactlyOnce handler branch
• Return ExactlyOnce handlers from stream, if enabled.