feat: customMessageType

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/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()
     };

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

@@ -5,8 +5,10 @@ class SignalParams extends Parameters {
   Keyset keyset;
   String channel;
   String payload;
+  String? customMessageType;
 
-  SignalParams(this.keyset, this.channel, this.payload);
+  SignalParams(this.keyset, this.channel, this.payload,
+      {this.customMessageType});
 
   @override
   Request toRequest() {
@@ -21,6 +23,7 @@ class SignalParams extends Parameters {
     ];
 
     var queryParameters = {
+      if (customMessageType != null) 'custom_message_type': customMessageType,
       if (keyset.authKey != null) 'auth': keyset.authKey,
       'uuid': keyset.uuid.value,
     };

pubnub/lib/src/dx/channel/channel.dart

@@ -41,13 +41,18 @@ class Channel {
   /// If set to `0`, message won't expire.
   /// If unset, expiration will fall back to default.
   Future<PublishResult> publish(dynamic message,
-      {bool? storeMessage, int? ttl, dynamic meta, bool? fire}) {
+      {bool? storeMessage,
+      int? ttl,
+      dynamic meta,
+      bool? fire,
+      String? customMessageType}) {
     return _core.publish(name, message,
         storeMessage: storeMessage,
         ttl: ttl,
         keyset: _keyset,
         meta: meta,
-        fire: fire);
+        fire: fire,
+        customMessageType: customMessageType);
   }
 
   /// Returns [PaginatedChannelHistory]. Most useful in infinite list type scenario.