feat: customMessageType (#128)

* * `custom_message_type` support for subscription, publish, signal and files features.
* `custom_message_type` field parsing support in history apis.
* acceptance tests steps implementation for custom_message_type

* set default retry policy to `Exponential` with default configuration.
Fixed issue pointed out by PR #127 for limiting retry delay within maximum delay for linear policy

* fix: linting and warnings

* fix channel test for customMessageType

* PubNub SDK v5.0.0 release.

* update changelog

---------

Co-authored-by: PubNub Release Bot <[email protected]>

.pubnub.yml

@@ -1,5 +1,14 @@
 ---
 changelog:
+  - date: 2024-12-10
+    version: v5.0.0
+    changes:
+      - type: feature
+        text: "BREAKING CHANGES: Default retry policy for subscription is set to Exponential."
+      - type: feature
+        text: "support for customMessageType in subscription, history, publish, signal and files features."
+      - type: bug
+        text: "Limiting delay to maximum allowable value for Linear retry policy."
   - date: 2024-04-15
     version: v4.3.4
     changes:
@@ -452,7 +461,7 @@ supported-platforms:
     platforms:
       - "Dart SDK >=2.6.0 <3.0.0"
     version: "PubNub Dart SDK"
-version: "4.3.4"
+version: "5.0.0"
 sdks:
   -
     full-name: PubNub Dart SDK

acceptance_tests/lib/src/steps/customMessageType/customMessageType_steps.dart

@@ -0,0 +1,24 @@
+import '../../world.dart';
+import 'package:gherkin/gherkin.dart';
+
+import 'step_given_keyset.dart';
+import 'step_then_error_response.dart';
+import 'step_then_messagesContainsType.dart';
+import 'step_then_receive.dart';
+import 'step_then_success_response.dart';
+import 'step_when_publish_with_type.dart';
+import 'step_when_sendFile.dart';
+import 'step_when_signal_with_type.dart';
+import 'step_when_subscribe.dart';
+
+final List<StepDefinitionGeneric<PubNubWorld>> customMessageTypeSteps = [
+  StepGivenTheDemoKeyset(),
+  StepWhenIPublishWithCustomType(),
+  StepThenIReceiveSuccessfulResponsePublish(),
+  StepThenIReceivePublishErrorResponse(),
+  StepWhenISignalWithCustomType(),
+  StepWhenISubscribeChannalForCustomMessageType(),
+  StepThenIReceiveMessagesInSubscriptionResponse(),
+  StepThenReceivedMessagesHasMessageTypes(),
+  StepWhenISendFileCustomType(),
+];

acceptance_tests/lib/src/steps/customMessageType/step_given_keyset.dart

@@ -0,0 +1,18 @@
+import 'package:gherkin/gherkin.dart';
+import 'package:pubnub/pubnub.dart';
+
+import '../../world.dart';
+
+class StepGivenTheDemoKeyset extends GivenWithWorld<PubNubWorld> {
+  @override
+  RegExp get pattern => RegExp(r'the demo keyset');
+
+  @override
+  Future<void> executeStep() async {
+    world.keyset = Keyset(
+      publishKey: 'demo',
+      subscribeKey: 'demo',
+      userId: UserId('testCustomType')      
+    );
+  }
+}

acceptance_tests/lib/src/steps/customMessageType/step_then_error_response.dart

@@ -0,0 +1,21 @@
+import 'package:gherkin/gherkin.dart';
+import 'package:pubnub/pubnub.dart';
+import 'package:test/test.dart';
+
+import '../../world.dart';
+
+class StepThenIReceivePublishErrorResponse extends ThenWithWorld<PubNubWorld> {
+  @override
+  RegExp get pattern => RegExp(r'I receive an error response');
+
+  @override
+  Future<void> executeStep() async {
+    if(world.latestResultType == 'sendFile') {
+      var result = world.latestResult as PublishFileMessageResult;
+      this.expect(result.description?.toLowerCase(), contains('invalid_type'));
+    } else {
+       var result = world.latestResult as PublishException;
+      this.expect(result.message.toLowerCase(), contains('invalid_type'));
+    }
+  }
+}

acceptance_tests/lib/src/steps/customMessageType/step_then_messagesContainsType.dart

@@ -0,0 +1,17 @@
+import 'package:gherkin/gherkin.dart';
+import 'package:test/expect.dart';
+
+import '../../world.dart';
+
+class StepThenReceivedMessagesHasMessageTypes
+    extends Then2WithWorld<String, String, PubNubWorld> {
+  @override
+  RegExp get pattern => RegExp(r'response contains messages with {string} and {string} types');
+
+  @override
+  Future<void> executeStep(String customMessageTypeOne, String customMessageTypeTwo) async {
+    world.messages.forEach((message) {
+      this.expect(message.customMessageType, anyOf([customMessageTypeOne, customMessageTypeTwo]));
+    });
+  }
+}

acceptance_tests/lib/src/steps/customMessageType/step_then_receive.dart

@@ -0,0 +1,14 @@
+import 'package:gherkin/gherkin.dart';
+
+import '../../world.dart';
+
+class StepThenIReceiveMessagesInSubscriptionResponse
+    extends Then1WithWorld<int, PubNubWorld> {
+  @override
+  RegExp get pattern => RegExp(r'I receive {int} messages in my subscribe response');
+
+  @override
+  Future<void> executeStep(int count) async {
+    expect(world.messages.length, 2);
+  }
+}

acceptance_tests/lib/src/steps/customMessageType/step_then_success_response.dart

@@ -0,0 +1,14 @@
+import 'package:gherkin/gherkin.dart';
+import 'package:test/expect.dart';
+
+import '../../world.dart';
+
+class StepThenIReceiveSuccessfulResponsePublish extends ThenWithWorld<PubNubWorld> {
+  @override
+  RegExp get pattern => RegExp(r'I receive a successful response');
+
+  @override
+  Future<void> executeStep() async {
+    this.expect(world.latestResultType, isNotNull);
+  }
+}

acceptance_tests/lib/src/steps/customMessageType/step_when_publish_with_type.dart

@@ -0,0 +1,24 @@
+import 'package:gherkin/gherkin.dart';
+
+import '../../world.dart';
+
+class StepWhenIPublishWithCustomType extends When1WithWorld<String, PubNubWorld> {
+  @override
+  RegExp get pattern => RegExp(r'I publish message with {string} customMessageType');
+
+  @override
+  Future<void> executeStep(String customMesageType) async {
+    try {
+      world.latestResult = await world.pubnub.publish(
+        'test',
+        'hello',
+        keyset: world.keyset,
+        customMessageType: customMesageType,
+      );
+      world.latestResultType = 'publish';
+    } catch (e) {
+      world.latestResultType = 'publishWithCustomTypeFailure';
+      world.latestResult = e;
+    }
+  }
+}

acceptance_tests/lib/src/steps/customMessageType/step_when_sendFile.dart

@@ -0,0 +1,21 @@
+import 'package:gherkin/gherkin.dart';
+
+import '../../world.dart';
+
+class StepWhenISendFileCustomType
+    extends When1WithWorld<String, PubNubWorld> {
+  @override
+  RegExp get pattern =>
+      RegExp(r'I send a file with {string} customMessageType');
+
+  @override
+  Future<void> executeStep(String customMesageType) async {
+    try {
+      world.latestResult = await world.pubnub.files.sendFile('test', 'helloFile.txt', [12,16], customMessageType: customMesageType);
+      world.latestResultType = 'sendFile';
+    } catch (e) {
+      world.latestResultType = 'sendFileFailure';
+      world.latestResult = e;
+    }
+  }
+}

acceptance_tests/lib/src/steps/customMessageType/step_when_signal_with_type.dart

@@ -0,0 +1,26 @@
+import 'package:gherkin/gherkin.dart';
+
+import '../../world.dart';
+
+class StepWhenISignalWithCustomType
+    extends When1WithWorld<String, PubNubWorld> {
+  @override
+  RegExp get pattern =>
+      RegExp(r'I send a signal with {string} customMessageType');
+
+  @override
+  Future<void> executeStep(String customMesageType) async {
+    try {
+      world.latestResult = await world.pubnub.signal(
+        'test',
+        'hello',
+        keyset: world.keyset,
+        customMessageType: customMesageType,
+      );
+      world.latestResultType = 'publish';
+    } catch (e) {
+      world.latestResultType = 'publishWithCustomTypeFailure';
+      world.latestResult = e;
+    }
+  }
+}

acceptance_tests/lib/src/steps/customMessageType/step_when_subscribe.dart

@@ -0,0 +1,27 @@
+import 'package:gherkin/gherkin.dart';
+
+import '../../world.dart';
+
+class StepWhenISubscribeChannalForCustomMessageType
+    extends When1WithWorld<String, PubNubWorld> {
+  @override
+  RegExp get pattern =>
+      RegExp(r'I subscribe to {string} channel');
+
+  @override
+  Future<void> executeStep(String channel) async {
+    try {
+      var subscription = world.pubnub.subscribe(channels: {channel});
+      subscription.messages.listen((messageEnvelope) {
+        world.messages.add(messageEnvelope);
+      });
+      await Future.delayed(Duration(seconds: 2), () {
+        subscription.dispose();
+      });
+      world.latestResultType = 'subscription';
+    } catch (e) {
+      world.latestResultType = 'subscriptionFailure';
+      world.latestResult = e;
+    }
+  }
+}

acceptance_tests/lib/src/steps/steps.dart

@@ -1,3 +1,4 @@
+import 'package:acceptance_tests/src/steps/customMessageType/customMessageType_steps.dart';
 import 'package:gherkin/gherkin.dart';
 
 import '../world.dart';
@@ -26,6 +27,7 @@ import 'steps_push.dart';
 final List<StepDefinitionGeneric<PubNubWorld>> steps = [
   ...cryptoSteps,
   ...pamv3Steps,
+  ...customMessageTypeSteps,
   StepGivenChannel(),
   StepGivenDemoKeyset(),
   StepWhenIAddAMessageAction(),

acceptance_tests/lib/src/world.dart

@@ -43,6 +43,7 @@ class PubNubWorld extends World {
       ),
       networking: NetworkingModule(origin: 'localhost:8090', ssl: false),
     );
+    pubnub.keysets.defaultKeyset.fileMessagePublishRetryLimit = 0;
   }
 
   Future<void> cleanup() async {

pubnub/CHANGELOG.md

@@ -1,3 +1,9 @@
+## v5.0.0
+December 10 2024
+
+#### Added
+- BREAKING CHANGES: support for customMessageType in subscription, history, publish, signal and files features.
+
 ## v4.3.4
 April 15 2024
 

pubnub/README.md

@@ -14,7 +14,7 @@ To add the package to your Dart or Flutter project, add `pubnub` as a dependency
 
 ```yaml
 dependencies:
-  pubnub: ^4.3.4
+  pubnub: ^5.0.0
 ```
 
 After adding the dependency to `pubspec.yaml`, run the `dart pub get` command in the root directory of your project (the same that the `pubspec.yaml` is in).

pubnub/lib/src/core/core.dart

@@ -21,7 +21,7 @@ class Core {
   /// Internal module responsible for supervising.
   SupervisorModule supervisor = SupervisorModule();
 
-  static String version = '4.3.4';
+  static String version = '5.0.0';
 
   Core(
       {Keyset? defaultKeyset,

pubnub/lib/src/core/policies/retry_policy.dart

@@ -36,12 +36,13 @@ class LinearRetryPolicy extends RetryPolicy {
   const LinearRetryPolicy({int? backoff, int? maxRetries, int? maximumDelay})
       : backoff = backoff ?? 5,
         maximumDelay = maximumDelay ?? 60000,
-        super(maxRetries ?? 5);
+        super(maxRetries ?? 10);
 
   @override
   Duration getDelay(Fiber fiber) {
     return Duration(
-        milliseconds: (fiber.tries * backoff) + Random().nextInt(1000));
+        milliseconds: min(
+            maximumDelay, (fiber.tries * backoff) + Random().nextInt(1000)));
   }
 }
 
@@ -53,8 +54,8 @@ class ExponentialRetryPolicy extends RetryPolicy {
   final int maximumDelay;
 
   const ExponentialRetryPolicy({int? maxRetries, int? maximumDelay})
-      : maximumDelay = maximumDelay ?? 60000,
-        super(maxRetries ?? 5);
+      : maximumDelay = maximumDelay ?? 150000,
+        super(maxRetries ?? 6);
 
   @override
   Duration getDelay(Fiber fiber) {

pubnub/lib/src/dx/_endpoints/files.dart

@@ -74,9 +74,11 @@ class PublishFileMessageParams extends Parameters {
   bool? storeMessage;
   int? ttl;
   String? meta;
+  String? customMessageType;
 
   PublishFileMessageParams(this.keyset, this.channel, this.message,
-      {this.storeMessage, this.meta, this.ttl});
+      {this.storeMessage, this.meta, this.ttl, this.customMessageType});
+
   @override
   Request toRequest() {
     var pathSegments = [
@@ -98,7 +100,8 @@ class PublishFileMessageParams extends Parameters {
         'store': '0',
       'uuid': keyset.uuid.value,
       if (ttl != null) 'ttl': ttl.toString(),
-      if (meta != null) 'meta': meta
+      if (meta != null) 'meta': meta,
+      if (customMessageType != null) 'custom_message_type': customMessageType,
     };
     return Request.get(
         uri: Uri(pathSegments: pathSegments, queryParameters: queryParameters));

pubnub/lib/src/dx/_endpoints/history.dart

@@ -158,8 +158,19 @@ class BatchHistoryResultEntry {
   /// for given `message`.
   PubNubException? error;
 
-  BatchHistoryResultEntry._(this.message, this.timetoken, this.uuid,
-      this.messageType, this.actions, this.meta, this.error);
+  /// If message has customMessageType, this will contain it.
+  /// Otherwise, it will be `null`.
+  String? customMessageType;
+
+  BatchHistoryResultEntry._(
+      this.message,
+      this.timetoken,
+      this.uuid,
+      this.messageType,
+      this.actions,
+      this.meta,
+      this.error,
+      this.customMessageType);
 
   /// @nodoc
   factory BatchHistoryResultEntry.fromJson(Map<String, dynamic> object,
@@ -195,7 +206,8 @@ class BatchHistoryResultEntry {
         MessageTypeExtension.fromInt(object['message_type']),
         object['actions'],
         object['meta'] == '' ? null : object['meta'],
-        error);
+        error,
+        object['custom_message_type']);
   }
 }
 

pubnub/lib/src/dx/_endpoints/publish.dart

@@ -9,9 +9,13 @@ class PublishParams extends Parameters {
   bool? storeMessage;
   int? ttl;
   bool? noReplication;
+  String? customMessageType;
 
   PublishParams(this.keyset, this.channel, this.message,
-      {this.storeMessage, this.ttl, this.noReplication});
+      {this.storeMessage,
+      this.ttl,
+      this.noReplication,
+      this.customMessageType});
 
   @override
   Request toRequest() {
@@ -33,6 +37,7 @@ class PublishParams extends Parameters {
       if (meta != null) 'meta': meta,
       if (noReplication != null && noReplication == true) 'norep': 'true',
       if (keyset.authKey != null) 'auth': keyset.authKey,
+      if (customMessageType != null) 'custom_message_type': customMessageType,
       'uuid': keyset.uuid.value,
       if (ttl != null) 'ttl': ttl.toString()
     };