[Async Attachment Upload] Initial implementation

Signed-off-by: Sahil Kumar <[email protected]>
2021-02-16 21:21:15 +05:30
parent 628542f99b
commit 0a6948cf32
54 changed files with 2561 additions and 1328 deletions
@@ -0,0 +1,81 @@
+import 'package:dio/dio.dart';
+import 'package:stream_chat/src/models/attachment_file.dart';
+import '../client.dart';
+import '../extensions/string_extension.dart';
+
+///
+abstract class AttachmentUploader {
+  ///
+  Future<String> uploadImage(
+    AttachmentFile file,
+    String channelId,
+    String channelType, {
+    ProgressCallback onSendProgress,
+    CancelToken cancelToken,
+  });
+
+  ///
+  Future<String> uploadFile(
+    AttachmentFile file,
+    String channelId,
+    String channelType, {
+    ProgressCallback onSendProgress,
+    CancelToken cancelToken,
+  });
+}
+
+///
+class StreamAttachmentUploader implements AttachmentUploader {
+  final StreamChatClient _client;
+
+  ///
+  const StreamAttachmentUploader(this._client);
+
+  @override
+  Future<String> uploadImage(
+    AttachmentFile file,
+    String channelId,
+    String channelType, {
+    ProgressCallback onSendProgress,
+    CancelToken cancelToken,
+  }) async {
+    final filename = file.path?.split('/')?.last;
+    final mimeType = filename.mimeType;
+    final res = await _client.sendImage(
+      await MultipartFile.fromFile(
+        file.path,
+        filename: filename,
+        contentType: mimeType,
+      ),
+      channelId,
+      channelType,
+      onSendProgress: onSendProgress,
+      cancelToken: cancelToken,
+    );
+    return res.file;
+  }
+
+  @override
+  Future<String> uploadFile(
+    AttachmentFile file,
+    String channelId,
+    String channelType, {
+    ProgressCallback onSendProgress,
+    CancelToken cancelToken,
+  }) async {
+    final filename = file.path?.split('/')?.last;
+    final mimeType = filename.mimeType;
+    final res = await _client.sendFile(
+      await MultipartFile.fromFile(
+        file.path,
+        filename: filename,
+        contentType: mimeType,
+      ),
+      channelId,
+      channelType,
+      onSendProgress: onSendProgress,
+      cancelToken: cancelToken,
+    );
+    return res.file;
+  }
+}
@@ -2,15 +2,16 @@ import 'dart:async';
 import 'dart:convert';
 import 'dart:math';

+import 'package:pedantic/pedantic.dart' show unawaited;
 import 'package:dio/dio.dart';
 import 'package:logging/logging.dart';
 import 'package:rxdart/rxdart.dart';
 import 'package:stream_chat/src/api/retry_queue.dart';
 import 'package:stream_chat/src/event_type.dart';
+import 'package:stream_chat/src/models/attachment_file.dart';
 import 'package:stream_chat/src/models/channel_state.dart';
 import 'package:stream_chat/src/models/user.dart';
 import 'package:stream_chat/stream_chat.dart';
-import 'package:uuid/uuid.dart';

 import '../client.dart';
 import '../models/event.dart';
@@ -171,19 +172,148 @@ class Channel {
  /// Call [watch] to initialize the client or instantiate it using [Channel.fromState]
  Future<bool> get initialized => _initializedCompleter.future;

-  /// Send a message to this channel
+  final _cancelableAttachmentUploadRequest = <String, CancelToken>{};
+  final _messageAttachmentsUploadCompleter = <String, Completer>{};
+
+  /// Cancels [attachmentId] upload request. Throws exception if the request hasn't
+  /// even started yet, Already completed or Already cancelled.
+  ///
+  /// Optionally, provide a [reason] for the cancellation.
+  void cancelAttachmentUpload(
+    String attachmentId, {
+    String reason,
+  }) {
+    final cancelToken = _cancelableAttachmentUploadRequest[attachmentId];
+    if (cancelToken == null) {
+      throw Exception(
+        "Upload request for this Attachment hasn't started yet or else Already completed",
+      );
+    }
+    if (cancelToken.isCancelled) throw Exception('Already cancelled');
+    cancelToken.cancel(reason);
+  }
+
+  /// Retries the failed [attachmentId] upload request.
+  Future<void> retryAttachmentUpload(String messageId, String attachmentId) {
+    return _uploadAttachments(messageId, [attachmentId]);
+  }
+
+  Future<void> _uploadAttachments(
+    String messageId,
+    Iterable<String> attachmentIds,
+  ) {
+    var message = state.messages.firstWhere(
+      (it) => it.id == messageId,
+      orElse: () => null,
+    );
+
+    if (message == null) {
+      throw Exception('Error, Message not found');
+    }
+
+    final attachments = message.attachments.where((it) {
+      if (it.uploadState.isSuccess) return false;
+      return attachmentIds.contains(it.id);
+    });
+
+    if (attachments.isEmpty) {
+      client.logger.info('No attachments available to upload');
+      if (message.attachments.every((it) => it.uploadState.isSuccess)) {
+        _messageAttachmentsUploadCompleter.remove(messageId)?.complete(message);
+      }
+      return Future.value();
+    }
+
+    client.logger.info('Found ${attachments.length} attachments');
+    return Future.wait(attachments.map((it) {
+      client.logger.info('Uploading ${it.id} attachment...');
+
+      void updateAttachment(Attachment attachment) {
+        message = message.copyWith(
+            attachments: message.attachments.map((it) {
+          if (it.id != attachment.id) return it;
+          return attachment;
+        }).toList(growable: false));
+        state?.addMessage(message);
+      }
+
+      void onSendProgress(int sent, int total) {
+        updateAttachment(it.copyWith(
+          uploadState: UploadState.inProgress(uploaded: sent, total: total),
+        ));
+      }
+
+      final isImage = it.type == 'image';
+      final uploader = _client.attachmentUploader;
+      final cancelToken = CancelToken();
+      Future<String> future;
+      if (isImage) {
+        future = uploader.uploadImage(it.file, id, type,
+            onSendProgress: onSendProgress, cancelToken: cancelToken);
+      } else {
+        future = uploader.uploadFile(it.file, id, type,
+            onSendProgress: onSendProgress, cancelToken: cancelToken);
+      }
+      _cancelableAttachmentUploadRequest[it.id] = cancelToken;
+      return future.then((url) {
+        client.logger.info('Attachment ${it.id} uploaded successfully...');
+        if (isImage) {
+          updateAttachment(
+            it.copyWith(imageUrl: url, uploadState: UploadState.success()),
+          );
+        } else {
+          updateAttachment(
+            it.copyWith(assetUrl: url, uploadState: UploadState.success()),
+          );
+        }
+      }).catchError((e, stk) {
+        updateAttachment(
+          it.copyWith(uploadState: UploadState.failed(error: e.toString())),
+        );
+      }).whenComplete(() {
+        _cancelableAttachmentUploadRequest.remove(it.id);
+      });
+    })).whenComplete(() {
+      if (message.attachments.every((it) => it.uploadState.isSuccess)) {
+        _messageAttachmentsUploadCompleter.remove(messageId)?.complete(message);
+      }
+    });
+  }
+
+  /// Send a [message] to this channel. Optionally pass a [attachmentUploader]
+  /// for custom attachments upload.
+  ///
+  /// Waits for a [_messageAttachmentsUploadCompleter] to complete
+  /// before actually sending the message.
  Future<SendMessageResponse> sendMessage(Message message) async {
-    final messageId = message.id ?? Uuid().v4();
+    // Cancelling previous completer in case it's called again in the process
+    // Eg. Updating the message while the previous call is in progress.
+    _messageAttachmentsUploadCompleter
+        .remove(message.id)
+        ?.completeError('Message Cancelled');
+
    final quotedMessage = state?.messages?.firstWhere(
      (m) => m.id == message?.quotedMessageId,
      orElse: () => null,
    );
-    final newMessage = message.copyWith(
+    message = message.copyWith(
      createdAt: message.createdAt ?? DateTime.now(),
      user: _client.state.user,
-      id: messageId,
      quotedMessage: quotedMessage,
      status: MessageSendingStatus.sending,
+      attachments: [
+        ...message.attachments.map(
+          (it) {
+            if (it.uploadState.isSuccess) return it;
+            return it.copyWith(
+              uploadState: UploadState.inProgress(
+                uploaded: 0,
+                total: it.file?.size ?? it.extraData['file_size'],
+              ),
+            );
+          },
+        )
+      ],
    );

    if (message.parentId != null && message.id == null) {
@@ -195,17 +325,24 @@ class Channel {
      ));
    }

-    state?.addMessage(newMessage);
+    state?.addMessage(message);

    try {
+      final attachmentsUploadCompleter = Completer<Message>();
+      _messageAttachmentsUploadCompleter[message.id] =
+          attachmentsUploadCompleter;
+
+      unawaited(_uploadAttachments(
+        message.id,
+        message.attachments.map((it) => it.id),
+      ));
+
+      message = await attachmentsUploadCompleter.future;
+
      final response = await _client.post(
        '$_channelURL/message',
        data: {
-          'message': message
-              .copyWith(
-                id: messageId,
-              )
-              .toJson()
+          'message': message.toJson(),
        },
      );

@@ -216,7 +353,104 @@ class Channel {
      return res;
    } catch (error) {
      if (error is DioError && error.type != DioErrorType.RESPONSE) {
-        state?.retryQueue?.add([newMessage]);
+        state?.retryQueue?.add([message]);
+      }
+      rethrow;
+    }
+  }
+
+  /// Updates the [message] in this channel. Optionally pass a [attachmentUploader]
+  /// for custom attachments upload.
+  ///
+  /// Waits for a [_messageAttachmentsUploadCompleter] to complete
+  /// before actually updating the message.
+  Future<UpdateMessageResponse> updateMessage(Message message) async {
+    // Cancelling previous completer in case it's called again in the process
+    // Eg. Updating the message while the previous call is in progress.
+    _messageAttachmentsUploadCompleter
+        .remove(message.id)
+        ?.completeError('Message Cancelled');
+
+    message = message.copyWith(
+      status: MessageSendingStatus.updating,
+      updatedAt: message.updatedAt ?? DateTime.now(),
+      attachments: [
+        ...message.attachments.map(
+          (it) {
+            if (it.uploadState.isSuccess) return it;
+            return it.copyWith(
+              uploadState: UploadState.inProgress(
+                uploaded: 0,
+                total: it.file?.size ?? it.extraData['file_size'],
+              ),
+            );
+          },
+        )
+      ],
+    );
+
+    state?.addMessage(message);
+
+    try {
+      final attachmentsUploadCompleter = Completer<Message>();
+      _messageAttachmentsUploadCompleter[message.id] =
+          attachmentsUploadCompleter;
+
+      unawaited(_uploadAttachments(
+        message.id,
+        message.attachments.map((it) => it.id),
+      ));
+
+      message = await attachmentsUploadCompleter.future;
+
+      final response = await _client.updateMessage(message);
+      state?.addMessage(response?.message?.copyWith(
+        ownReactions: message.ownReactions,
+      ));
+      return response;
+    } catch (error) {
+      if (error is DioError && error.type != DioErrorType.RESPONSE) {
+        state?.retryQueue?.add([message]);
+      }
+      rethrow;
+    }
+  }
+
+  /// Deletes the [message] from the channel.
+  Future<EmptyResponse> deleteMessage(Message message) async {
+    // Directly deleting the local messages which are not yet sent to server
+    if (message.status == MessageSendingStatus.sending ||
+        message.status == MessageSendingStatus.failed) {
+      state.addMessage(message.copyWith(
+        type: 'deleted',
+        status: MessageSendingStatus.sent,
+      ));
+
+      // Removing the attachments upload completer to stop the `sendMessage`
+      // waiting for attachments to complete.
+      _messageAttachmentsUploadCompleter
+          .remove(message.id)
+          ?.completeError(Exception('Message deleted'));
+      return EmptyResponse();
+    }
+
+    try {
+      message = message.copyWith(
+        type: 'deleted',
+        status: MessageSendingStatus.deleting,
+        deletedAt: message.deletedAt ?? DateTime.now(),
+      );
+
+      state?.addMessage(message);
+
+      final response = await _client.deleteMessage(message);
+
+      state?.addMessage(message.copyWith(status: MessageSendingStatus.sent));
+
+      return response;
+    } catch (error) {
+      if (error is DioError && error.type != DioErrorType.RESPONSE) {
+        state?.retryQueue?.add([message]);
      }
      rethrow;
    }
@@ -226,26 +460,30 @@ class Channel {
  Future<SendFileResponse> sendFile(
    MultipartFile file, {
    ProgressCallback onSendProgress,
-  }) async {
-    final response = await _client.post(
-      '$_channelURL/file',
-      data: FormData.fromMap({'file': file}),
+    CancelToken cancelToken,
+  }) {
+    return _client.sendFile(
+      file,
+      id,
+      type,
      onSendProgress: onSendProgress,
+      cancelToken: cancelToken,
    );
-    return _client.decode(response.data, SendFileResponse.fromJson);
  }

  /// Send an image to this channel
  Future<SendImageResponse> sendImage(
-    MultipartFile file, {
+    MultipartFile image, {
    ProgressCallback onSendProgress,
-  }) async {
-    final response = await _client.post(
-      '$_channelURL/image',
-      data: FormData.fromMap({'file': file}),
+    CancelToken cancelToken,
+  }) {
+    return _client.sendImage(
+      image,
+      id,
+      type,
      onSendProgress: onSendProgress,
+      cancelToken: cancelToken,
    );
-    return _client.decode(response.data, SendImageResponse.fromJson);
  }

  /// Delete a file from this channel
@@ -975,7 +1213,13 @@ class ChannelClientState {
        ?.getChannelThreads(_channel.cid)
        ?.then((threads) {
      _threads = threads;
-      retryFailedMessages();
+    })?.then((_) {
+      _channel._client.chatPersistenceClient
+          ?.getChannelStateByCid(_channel.cid)
+          ?.then((state) {
+        updateChannelState(state);
+        retryFailedMessages();
+      });
    });
  }

@@ -125,10 +125,7 @@ class RetryQueue {
  Future<void> _sendMessage(Message message) async {
    if (message.status == MessageSendingStatus.failed_update ||
        message.status == MessageSendingStatus.updating) {
-      await channel.client.updateMessage(
-        message,
-        channel.cid,
-      );
+      await channel.updateMessage(message);
    } else if (message.status == MessageSendingStatus.failed ||
        message.status == MessageSendingStatus.sending) {
      await channel.sendMessage(
@@ -136,10 +133,7 @@ class RetryQueue {
      );
    } else if (message.status == MessageSendingStatus.failed_delete ||
        message.status == MessageSendingStatus.deleting) {
-      await channel.client.deleteMessage(
-        message,
-        channel.cid,
-      );
+      await channel.client.deleteMessage(message);
    }
  }