talksik/stream-chat-flutter
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Release/1.4.0 beta (#318)

Browse Source 
* fix: fixed long system messages overflowing

* fix: fixed long system messages overflowing

* [UI-Kit] Refactor, add support for downloading attachments.

Signed-off-by: Sahil Kumar <[email protected]>

* fix: header casing

* [UI-Kit] Disable attachment download dialog pop using back button

Signed-off-by: Sahil Kumar <[email protected]>

* stream_chat_flutter fix svg on web

* [UI-Kit] Pop attachment download dialog in case of any error.

Signed-off-by: Sahil Kumar <[email protected]>

* restore svg

* feat: add pr title linter

* fix test

* fix client headers

* update system message design

* update gh token env var

* add scopes

* expose systemMessageBuilder in messagelistview

* fix(ui): fix button size

* add dispatch workflow

* add dispatch workflow

* update dispatch acton

* add runson

* fix ref

* fix ref

* update secret

* specify repo

* fix: list scroll keyboard behaviour  (#304)

* fix: Keyboard now closes by clicking on space or scrolling.

* fix: Exposed system and normal message tap builders

* fmt: dartfmt

* feat: Add support for messages filter in `MessageListView` and `MessageListCore` (#303)

* [MessageListView, MessageListCore] Add support for message filter

Signed-off-by: Sahil Kumar <[email protected]>

* [UI-Kit -> Pubspec] Remove `stream_chat_flutter_core` relative import

Signed-off-by: Sahil Kumar <[email protected]>

* add relative import

* add relative import

* remove svg for web

* remove badge logic on web

* style: Update for team lint (#297)

* Update for team lint

* fix tests

* remove pedantic and sort deps

* remove jiffy from system message

Co-authored-by: Salvatore Giordano <[email protected]>

* add section highlighting sample app repo (#307)

Co-authored-by: Salvatore Giordano <[email protected]>

* add null check to debounce

* fix messagelistview loading errors

* remove core relative import

* fix: fixes message newline issue (#308)

* fix: minor

* fix: generate image thumbnails and use transparent image as placeholder for cached network images (#310)

* update cdn handling

* fix image attachment with new cdn

* fix(ui): Fix message input permission request crash (#311)

Signed-off-by: Sahil Kumar <[email protected]>

* fix(llc): update member presence and add skip_push to message (#314)

* update member presence

* add skip_push to message

* fix tests

* fix: Fixed shimmer overflow (#316)

* fix dialogs

* fix(llc): Fix attachment upload state uneven progress. (#315)

* feat(llc): Add rate limiter functions.
fix(llc): Fix attachment upload state uneven progress.

Signed-off-by: Sahil Kumar <[email protected]>

* feat(llc): Export `async.dart` and `rate_limit.dart`

Signed-off-by: Sahil Kumar <[email protected]>

* refactor(llc): Replace our `rate_limit` implementation with the dart port of js `lodash` to better handlw `cancel` and `flush` functions.

Signed-off-by: Sahil Kumar <[email protected]>

* fix(UI-Kit): Make android example run on real devices

Signed-off-by: Sahil Kumar <[email protected]>

* fix offline behaviour

Co-authored-by: Salvatore Giordano <[email protected]>

* feat: Introduce onAttachmentTap on messages attachment (#309)

* introdce onAttachmentTap on message

* correct indent

* sdfsd

Co-authored-by: Salvatore Giordano <[email protected]>

* feat: added customization options in main widgets (#312)

* polish channelpreview customization options

* polish channelheader customization options

* polish messageinput customization options

* polish threadheader customization options

* polish channellistheader customization options

* show the parent message if no messages in thread

* fix edit message

* fix channelinfo textstyle

* fix review

* fix useravatar

* add ontitle tap to thread header

* add custom message actions

* add borderradius and button builder

* add ontap to messageinput custom send button

* fix bottom sheet

* add doc

* use placeholder image

* add doc

* fix listtile density

* remove subtitle from ChannelListHeaderTheme.copyWith

* add InputDecoration.merge extension

* fix: use shimmer in images

* fix: remove notification badge logic from the sdk and move it to sample app

* fix import

* fix: Method being called during build (#317)

* method call bug fix

* dartfmt

* fix: show error messages as system and keep them in the message input (#319)

* fix: pre release (#324)

* use share_plus plugin

* bump dependency versions

* fix ui for web

* unfocus messageinput only on commands

* fix default error for messagesearchlistview

* fix thumbnail animation

* fix lint

* ignore example in linter

* update changelogs

* fix lint

* fix lint

* update stream_chat pana min

* add doc

Co-authored-by: Sahil Kumar <[email protected]>
Co-authored-by: Deven Joshi <[email protected]>
Co-authored-by: Neevash Ramdial (Nash) <[email protected]>
Co-authored-by: Nelson Nunes <[email protected]>
This commit is contained in:
Salvatore Giordano
2021-03-09 18:58:23 +01:00
committed by GitHub
co-authored by Sahil Kumar Deven Joshi Neevash Ramdial Nelson Nunes
parent 3332a48a0e
commit 4488fa6c3a
121 changed files with 3768 additions and 2376 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/stream_chat/lib/src/api/channel.dart
+235 -190
View File
@@ -2,24 +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/debounce.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 '../client.dart';
import '../models/event.dart';
import '../models/member.dart';
import '../models/message.dart';
import 'requests.dart';
import 'responses.dart';
import 'package:stream_chat/src/extensions/rate_limit.dart';
/// This a the class that manages a specific channel.
class Channel {
@@ -57,7 +49,8 @@ class Channel {
set extraData(Map<String, dynamic> extraData) {
if (_initializedCompleter.isCompleted) {
throw Exception(
'Once the channel is initialized you should use channel.update to update channel data');
'Once the channel is initialized you should use channel.update '
'to update channel data');
}
_extraData = extraData;
}
@@ -168,14 +161,15 @@ class Channel {
final Completer<bool> _initializedCompleter = Completer();
/// True if this is initialized
/// Call [watch] to initialize the client or instantiate it using [Channel.fromState]
/// Call [watch] to initialize the client or instantiate it using
/// [Channel.fromState]
Future<bool> get initialized => _initializedCompleter.future;
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.
/// 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(
@@ -185,7 +179,8 @@ class Channel {
final cancelToken = _cancelableAttachmentUploadRequest[attachmentId];
if (cancelToken == null) {
throw Exception(
"Upload request for this Attachment hasn't started yet or else Already completed",
"Upload request for this Attachment hasn't started yet or else "
'Already completed',
);
}
if (cancelToken.isCancelled) throw Exception('Already cancelled');
@@ -193,15 +188,14 @@ class Channel {
}
/// Retries the failed [attachmentId] upload request.
Future<void> retryAttachmentUpload(String messageId, String attachmentId) {
return _uploadAttachments(messageId, [attachmentId]);
}
Future<void> retryAttachmentUpload(String messageId, String attachmentId) =>
_uploadAttachments(messageId, [attachmentId]);
Future<void> _uploadAttachments(
String messageId,
Iterable<String> attachmentIds,
) {
var message = state.messages.firstWhere(
final message = state.messages.firstWhere(
(it) => it.id == messageId,
orElse: () => null,
);
@@ -224,29 +218,29 @@ class Channel {
}
client.logger.info('Found ${attachments.length} attachments');
void updateAttachment(Attachment attachment) {
final index =
message.attachments.indexWhere((it) => it.id == attachment.id);
if (index != -1) {
message.attachments[index] = attachment;
state?.addMessage(message);
}
}
return Future.wait(attachments.map((it) {
client.logger.info('Uploading ${it.id} attachment...');
void updateAttachment(Attachment attachment) {
final index = message.attachments.indexWhere((it) {
return it.id == attachment.id;
});
if (index != -1) {
message.attachments[index] = attachment;
state?.addMessage(message);
}
}
final throttledUpdateAttachment = updateAttachment.throttled(
const Duration(milliseconds: 500),
);
void onSendProgress(int sent, int total) {
debounce(
timeout: Duration(seconds: 1),
target: updateAttachment,
positionalArguments: [
it.copyWith(
uploadState: UploadState.inProgress(uploaded: sent, total: total),
),
],
);
throttledUpdateAttachment([
it.copyWith(
uploadState: UploadState.inProgress(uploaded: sent, total: total),
),
]);
}
final isImage = it.type == 'image';
@@ -270,18 +264,26 @@ class Channel {
client.logger.info('Attachment ${it.id} uploaded successfully...');
if (isImage) {
updateAttachment(
it.copyWith(imageUrl: url, uploadState: UploadState.success()),
it.copyWith(
imageUrl: url,
uploadState: const UploadState.success(),
),
);
} else {
updateAttachment(
it.copyWith(assetUrl: url, uploadState: UploadState.success()),
it.copyWith(
assetUrl: url,
uploadState: const UploadState.success(),
),
);
}
}).catchError((e, stk) {
client.logger.severe('error uploading the attachment', e, stk);
updateAttachment(
it.copyWith(uploadState: UploadState.failed(error: e.toString())),
);
}).whenComplete(() {
throttledUpdateAttachment?.cancel();
_cancelableAttachmentUploadRequest.remove(it.id);
});
})).whenComplete(() {
@@ -305,6 +307,7 @@ class Channel {
(m) => m.id == message?.quotedMessageId,
orElse: () => null,
);
// ignore: parameter_assignments
message = message.copyWith(
createdAt: message.createdAt ?? DateTime.now(),
user: _client.state.user,
@@ -313,7 +316,7 @@ class Channel {
attachments: message.attachments?.map(
(it) {
if (it.uploadState.isSuccess) return it;
return it.copyWith(uploadState: UploadState.preparing());
return it.copyWith(uploadState: const UploadState.preparing());
},
)?.toList(),
);
@@ -335,11 +338,13 @@ class Channel {
_messageAttachmentsUploadCompleter[message.id] =
attachmentsUploadCompleter;
unawaited(_uploadAttachments(
// ignore: unawaited_futures
_uploadAttachments(
message.id,
message.attachments.map((it) => it.id),
));
);
// ignore: parameter_assignments
message = await attachmentsUploadCompleter.future;
}
@@ -364,13 +369,14 @@ class Channel {
.remove(message.id)
?.completeError('Message Cancelled');
// ignore: parameter_assignments
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.preparing());
return it.copyWith(uploadState: const UploadState.preparing());
},
)?.toList(),
);
@@ -383,11 +389,13 @@ class Channel {
_messageAttachmentsUploadCompleter[message.id] =
attachmentsUploadCompleter;
unawaited(_uploadAttachments(
// ignore: unawaited_futures
_uploadAttachments(
message.id,
message.attachments.map((it) => it.id),
));
);
// ignore: parameter_assignments
message = await attachmentsUploadCompleter.future;
}
@@ -423,6 +431,7 @@ class Channel {
}
try {
// ignore: parameter_assignments
message = message.copyWith(
type: 'deleted',
status: MessageSendingStatus.deleting,
@@ -456,7 +465,7 @@ class Channel {
throw ArgumentError('Invalid timeout or Expiration date');
}
return true;
}());
}(), 'Check for invalid token or expiration date');
DateTime pinExpires;
if (timeoutOrExpirationDate is DateTime) {
@@ -475,39 +484,36 @@ class Channel {
}
/// Unpins provided message
Future<UpdateMessageResponse> unpinMessage(Message message) {
return updateMessage(message.copyWith(pinned: false));
}
Future<UpdateMessageResponse> unpinMessage(Message message) =>
updateMessage(message.copyWith(pinned: false));
/// Send a file to this channel
Future<SendFileResponse> sendFile(
AttachmentFile file, {
ProgressCallback onSendProgress,
CancelToken cancelToken,
}) {
return _client.sendFile(
file,
id,
type,
onSendProgress: onSendProgress,
cancelToken: cancelToken,
);
}
}) =>
_client.sendFile(
file,
id,
type,
onSendProgress: onSendProgress,
cancelToken: cancelToken,
);
/// Send an image to this channel
Future<SendImageResponse> sendImage(
AttachmentFile file, {
ProgressCallback onSendProgress,
CancelToken cancelToken,
}) {
return _client.sendImage(
file,
id,
type,
onSendProgress: onSendProgress,
cancelToken: cancelToken,
);
}
}) =>
_client.sendImage(
file,
id,
type,
onSendProgress: onSendProgress,
cancelToken: cancelToken,
);
/// A message search.
Future<SearchMessagesResponse> search({
@@ -515,35 +521,32 @@ class Channel {
Map<String, dynamic> messageFilters,
List<SortOption> sort,
PaginationParams paginationParams,
}) {
return _client.search(
{
'cid': {
r'$in': [cid],
}) =>
_client.search(
{
'cid': {
r'$in': [cid],
},
},
},
sort: sort,
query: query,
paginationParams: paginationParams,
messageFilters: messageFilters,
);
}
sort: sort,
query: query,
paginationParams: paginationParams,
messageFilters: messageFilters,
);
/// Delete a file from this channel
Future<EmptyResponse> deleteFile(
String url, {
CancelToken cancelToken,
}) {
return _client.deleteFile(url, id, type, cancelToken: cancelToken);
}
}) =>
_client.deleteFile(url, id, type, cancelToken: cancelToken);
/// Delete an image from this channel
Future<EmptyResponse> deleteImage(
String url, {
CancelToken cancelToken,
}) {
return _client.deleteImage(url, id, type, cancelToken: cancelToken);
}
}) =>
_client.deleteImage(url, id, type, cancelToken: cancelToken);
/// Send an event on this channel
Future<EmptyResponse> sendEvent(Event event) {
@@ -551,9 +554,7 @@ class Channel {
return _client.post(
'$_channelURL/event',
data: {'event': event.toJson()},
).then((res) {
return _client.decode(res.data, EmptyResponse.fromJson);
});
).then((res) => _client.decode(res.data, EmptyResponse.fromJson));
}
/// Send a reaction to this channel
@@ -643,11 +644,10 @@ class Channel {
}
final latestReactions = [...message.latestReactions ?? <Reaction>[]]
..removeWhere((r) {
return r.userId == reaction.userId &&
r.type == reaction.type &&
r.messageId == reaction.messageId;
});
..removeWhere((r) =>
r.userId == reaction.userId &&
r.type == reaction.type &&
r.messageId == reaction.messageId);
final ownReactions = [...latestReactions ?? <Reaction>[]]
..removeWhere((it) => it.userId != user.id);
@@ -825,7 +825,7 @@ class Channel {
})
..addAll(options);
var response;
ChannelState response;
try {
response = await query(options: watchOptions);
@@ -861,7 +861,8 @@ class Channel {
}
/// List the message replies for a parent message
/// Set [preferOffline] to true to avoid the api call if the data is already in the offline storage
/// Set [preferOffline] to true to avoid the api call if the data is already
/// in the offline storage
Future<QueryRepliesResponse> getReplies(
String parentId,
PaginationParams options, {
@@ -940,16 +941,15 @@ class Channel {
}
/// Creates a new channel
Future<ChannelState> create() async {
return query(options: {
'watch': false,
'state': false,
'presence': false,
});
}
Future<ChannelState> create() async => query(options: {
'watch': false,
'state': false,
'presence': false,
});
/// Query the API, get messages, members or other channel fields
/// Set [preferOffline] to true to avoid the api call if the data is already in the offline storage
/// Set [preferOffline] to true to avoid the api call if the data is already
/// in the offline storage
Future<ChannelState> query({
Map<String, dynamic> options = const {},
PaginationParams messagesPagination,
@@ -1112,8 +1112,9 @@ class Channel {
});
}
/// Hides the channel from [StreamChatClient.queryChannels] for the user until a message is added
/// If [clearHistory] is set to true - all messages will be removed for the user
/// Hides the channel from [StreamChatClient.queryChannels] for the user
/// until a message is added If [clearHistory] is set to true - all messages
/// will be removed for the user
Future<EmptyResponse> hide({bool clearHistory = false}) async {
_checkInitialized();
final response = await _client
@@ -1134,28 +1135,28 @@ class Channel {
return _client.decode(response.data, EmptyResponse.fromJson);
}
/// Stream of [Event] coming from websocket connection specific for the channel
/// Pass an eventType as parameter in order to filter just a type of event
/// Stream of [Event] coming from websocket connection specific for the
/// channel. Pass an eventType as parameter in order to filter just a type
/// of event
Stream<Event> on([
String eventType,
String eventType2,
String eventType3,
String eventType4,
]) {
return _client
.on(
eventType,
eventType2,
eventType3,
eventType4,
)
.where((e) => e.cid == cid);
}
]) =>
_client
.on(
eventType,
eventType2,
eventType3,
eventType4,
)
.where((e) => e.cid == cid);
DateTime _lastTypingEvent;
/// First of the [EventType.typingStart] and [EventType.typingStop] events based on the users keystrokes.
/// Call this on every keystroke.
/// First of the [EventType.typingStart] and [EventType.typingStop] events
/// based on the users keystrokes. Call this on every keystroke.
Future<void> keyStroke([String parentId]) async {
if (config?.typingEvents == false) {
return;
@@ -1196,17 +1197,21 @@ class Channel {
void _checkInitialized() {
if (!_initializedCompleter.isCompleted) {
throw Exception(
"Channel $cid hasn't been initialized yet. Make sure to call .watch() or to instantiate the client using [Channel.fromState]");
"Channel $cid hasn't been initialized yet. Make sure to call .watch()"
' or to instantiate the client using [Channel.fromState]');
}
}
}
/// The class that handles the state of the channel listening to the events
class ChannelClientState {
final _subscriptions = <StreamSubscription>[];
/// Creates a new instance listening to events and updating the state
ChannelClientState(this._channel, ChannelState channelState) {
ChannelClientState(
this._channel,
ChannelState channelState,
) : _debouncedUpdatePersistenceChannelState = _channel
?._client?.chatPersistenceClient?.updateChannelState
?.debounced(const Duration(seconds: 1)) {
retryQueue = RetryQueue(
channel: _channel,
logger: Logger('RETRY QUEUE ${_channel.cid}'),
@@ -1252,14 +1257,16 @@ class ChannelClientState {
_channel._client.chatPersistenceClient
?.getChannelStateByCid(_channel.cid)
?.then((state) {
// Replacing the persistence state members with the latest `channelState.members`
// as they may have changes over the time.
// Replacing the persistence state members with the latest
// `channelState.members` as they may have changes over the time.
updateChannelState(state.copyWith(members: channelState.members));
retryFailedMessages();
});
});
}
final _subscriptions = <StreamSubscription>[];
void _computeInitialUnread() {
final userRead = channelState?.read?.firstWhere(
(r) => r.user.id == _channel._client.state?.user?.id,
@@ -1277,11 +1284,16 @@ class ChannelClientState {
m.attachments?.isNotEmpty == true &&
m.attachments?.any((e) {
final url = e.imageUrl ?? e.assetUrl;
if (url == null || !url.contains('stream-io-cdn.com')) {
if (url == null || !url.contains('')) {
return false;
}
final uri = Uri.parse(url);
if (uri.host != 'stream-io-cdn.com' ||
uri.queryParameters['Expires'] == null) {
return false;
}
final expiration =
DateTime.parse(Uri.parse(url).queryParameters['Expires']);
DateTime.parse(uri.queryParameters['Expires']);
return expiration.isBefore(DateTime.now());
}) ==
true)
@@ -1338,8 +1350,8 @@ class ChannelClientState {
/// Flag which indicates if [ChannelClientState] contain latest/recent messages or not.
/// This flag should be managed by UI sdks.
/// When false, any new message (received by WebSocket event - [EventType.messageNew]) will not
/// be pushed on to message list.
/// When false, any new message (received by WebSocket event
/// - [EventType.messageNew]) will not be pushed on to message list.
bool get isUpToDate => _isUpToDateController.value;
set isUpToDate(bool isUpToDate) => _isUpToDateController.add(isUpToDate);
@@ -1357,12 +1369,18 @@ class ChannelClientState {
Future<void> retryFailedMessages() async {
final failedMessages =
<Message>[...messages, ...threads.values.expand((v) => v)]
.where((message) =>
message.status != null &&
message.status != MessageSendingStatus.sent &&
message.createdAt.isBefore(DateTime.now().subtract(Duration(
seconds: 1,
))))
.where(
(message) =>
message.status != null &&
message.status != MessageSendingStatus.sent &&
message.createdAt.isBefore(
DateTime.now().subtract(
const Duration(
seconds: 1,
),
),
),
)
.toList();
retryQueue.add(failedMessages);
@@ -1471,27 +1489,32 @@ class ChannelClientState {
return;
}
_subscriptions.add(_channel
.on(
EventType.messageRead,
EventType.notificationMarkRead,
)
.listen((event) {
final readList = List<Read>.from(_channelState?.read ?? []);
final userReadIndex = read?.indexWhere((r) => r.user.id == event.user.id);
_subscriptions.add(
_channel
.on(
EventType.messageRead,
EventType.notificationMarkRead,
)
.listen(
(event) {
final readList = List<Read>.from(_channelState?.read ?? []);
final userReadIndex =
read?.indexWhere((r) => r.user.id == event.user.id);
if (userReadIndex != null && userReadIndex != -1) {
final userRead = readList.removeAt(userReadIndex);
if (userRead.user?.id == _channel._client.state.user.id) {
_unreadCountController.add(0);
}
readList.add(Read(
user: event.user,
lastRead: event.createdAt,
));
_channelState = _channelState.copyWith(read: readList);
}
}));
if (userReadIndex != null && userReadIndex != -1) {
final userRead = readList.removeAt(userReadIndex);
if (userRead.user?.id == _channel._client.state.user.id) {
_unreadCountController.add(0);
}
readList.add(Read(
user: event.user,
lastRead: event.createdAt,
));
_channelState = _channelState.copyWith(read: readList);
}
},
),
);
}
/// Channel message list
@@ -1527,11 +1550,8 @@ class ChannelClientState {
List<Member>, Map<String, User>, List<Member>>(
channelStateStream.map((cs) => cs.members),
_channel.client.state.usersStream,
(members, users) {
return members
.map((e) => e.copyWith(user: users[e.user.id]))
.toList();
},
(members, users) =>
members.map((e) => e.copyWith(user: users[e.user.id])).toList(),
);
/// Channel watcher count
@@ -1551,9 +1571,7 @@ class ChannelClientState {
CombineLatestStream.combine2<List<User>, Map<String, User>, List<User>>(
channelStateStream.map((cs) => cs.watchers),
_channel.client.state.usersStream,
(watchers, users) {
return watchers.map((e) => users[e.id] ?? e).toList();
},
(watchers, users) => watchers.map((e) => users[e.id] ?? e).toList(),
);
/// Channel read list
@@ -1625,9 +1643,7 @@ class ChannelClientState {
true)
?.toList() ??
[],
];
newMessages.sort(_sortByCreatedAt);
]..sort(_sortByCreatedAt);
final newWatchers = <User>[
...updatedState?.watchers ?? [],
@@ -1690,16 +1706,11 @@ class ChannelClientState {
ChannelState get channelState => _channelStateController.value;
BehaviorSubject<ChannelState> _channelStateController;
final Debounce _debouncedUpdatePersistenceChannelState;
set _channelState(ChannelState v) {
_channelStateController.add(v);
if (_channel._client.persistenceEnabled) {
debounce(
timeout: Duration(milliseconds: 500),
target: _channel._client.chatPersistenceClient?.updateChannelState,
positionalArguments: [v],
);
}
_debouncedUpdatePersistenceChannelState?.call([v]);
}
/// The channel threads related to this channel
@@ -1735,28 +1746,62 @@ class ChannelClientState {
return;
}
_subscriptions.add(_channel.on(EventType.typingStart).listen((event) {
if (event.user.id != _channel.client.state.user.id) {
_typings[event.user] = DateTime.now();
_typingEventsController.add(_typings.keys.toList());
}
}));
_subscriptions.add(_channel.on(EventType.typingStop).listen((event) {
if (event.user.id != _channel.client.state.user.id) {
_typings.remove(event.user);
_typingEventsController.add(_typings.keys.toList());
}
}));
_subscriptions
..add(
_channel.on(EventType.typingStart).listen(
(event) {
if (event.user.id != _channel.client.state.user.id) {
_typings[event.user] = DateTime.now();
_typingEventsController.add(_typings.keys.toList());
}
},
),
)
..add(
_channel.on(EventType.typingStop).listen(
(event) {
if (event.user.id != _channel.client.state.user.id) {
_typings.remove(event.user);
_typingEventsController.add(_typings.keys.toList());
}
},
),
)
..add(
_channel
.on()
.where((event) =>
event.user != null &&
members?.any((m) => m.userId == event.user.id) == true)
.listen(
(event) {
final newMembers = List<Member>.from(members);
final oldMemberIndex =
newMembers.indexWhere((m) => m.userId == event.user.id);
if (oldMemberIndex > -1) {
final oldMember = newMembers.removeAt(oldMemberIndex);
updateChannelState(ChannelState(
members: [
...newMembers,
oldMember.copyWith(
user: event.user,
),
],
));
}
},
),
);
}
Timer _cleaningTimer;
void _startCleaning() {
if (_channel.config?.typingEvents == false) {
return;
}
_cleaningTimer = Timer.periodic(Duration(seconds: 1), (_) {
_cleaningTimer = Timer.periodic(const Duration(seconds: 1), (_) {
final now = DateTime.now();
if (_channel._lastTypingEvent != null &&
@@ -1769,8 +1814,9 @@ class ChannelClientState {
}
Timer _pinnedMessagesTimer;
void _startCleaningPinnedMessages() {
_pinnedMessagesTimer = Timer.periodic(Duration(seconds: 30), (_) {
_pinnedMessagesTimer = Timer.periodic(const Duration(seconds: 30), (_) {
final now = DateTime.now();
var expiredMessages = channelState.pinnedMessages
?.where((m) => m.pinExpires?.isBefore(now) == true)
@@ -1781,8 +1827,6 @@ class ChannelClientState {
.map((m) => m.copyWith(
pinExpires: null,
pinned: false,
pinnedAt: null,
pinnedBy: null,
))
.toList();
@@ -1811,6 +1855,7 @@ class ChannelClientState {
/// Call this method to dispose this object
void dispose() {
_debouncedUpdatePersistenceChannelState?.cancel();
_unreadCountController.close();
retryQueue.dispose();
_subscriptions.forEach((s) => s.cancel());
packages/stream_chat/lib/src/api/requests.dart
+37 -35
View File
@@ -5,22 +5,6 @@ part 'requests.g.dart';
/// Sorting options
@JsonSerializable(createFactory: false)
class SortOption<T> {
/// Ascending order
static const ASC = 1;
/// Descending order
static const DESC = -1;
/// A sorting field name
final String field;
/// A sorting direction
final int direction;
/// Sorting field Comparator required for offline sorting
@JsonKey(ignore: true)
final Comparator<T> comparator;
/// Creates a new SortOption instance
///
/// For example:
@@ -34,6 +18,24 @@ class SortOption<T> {
this.comparator,
});
/// Ascending order
// ignore: constant_identifier_names
static const ASC = 1;
/// Descending order
// ignore: constant_identifier_names
static const DESC = -1;
/// A sorting field name
final String field;
/// A sorting direction
final int direction;
/// Sorting field Comparator required for offline sorting
@JsonKey(ignore: true)
final Comparator<T> comparator;
/// Serialize model to json
Map<String, dynamic> toJson() => _$SortOptionToJson(this);
}
@@ -41,6 +43,25 @@ class SortOption<T> {
/// Pagination options.
@JsonSerializable(createFactory: false, includeIfNull: false)
class PaginationParams {
/// Creates a new PaginationParams instance
///
/// For example:
/// ```dart
/// // limit to 50
/// final paginationParams = PaginationParams(limit: 50);
///
/// // limit to 50 with offset
/// final paginationParams = PaginationParams(limit: 50, offset: 50);
/// ```
const PaginationParams({
this.limit = 10,
this.offset = 0,
this.greaterThan,
this.greaterThanOrEqual,
this.lessThan,
this.lessThanOrEqual,
});
/// The amount of items requested from the APIs.
final int limit;
@@ -63,25 +84,6 @@ class PaginationParams {
@JsonKey(name: 'id_lte')
final String lessThanOrEqual;
/// Creates a new PaginationParams instance
///
/// For example:
/// ```dart
/// // limit to 50
/// final paginationParams = PaginationParams(limit: 50);
///
/// // limit to 50 with offset
/// final paginationParams = PaginationParams(limit: 50, offset: 50);
/// ```
const PaginationParams({
this.limit = 10,
this.offset = 0,
this.greaterThan,
this.greaterThanOrEqual,
this.lessThan,
this.lessThanOrEqual,
});
/// Serialize model to json
Map<String, dynamic> toJson() => _$PaginationParamsToJson(this);
packages/stream_chat/lib/src/api/requests.g.dart
+1 -1
View File
@@ -6,7 +6,7 @@ part of 'requests.dart';
// JsonSerializableGenerator
// **************************************************************************
Map<String, dynamic> _$SortOptionToJson(SortOption instance) =>
Map<String, dynamic> _$SortOptionToJson<T>(SortOption<T> instance) =>
<String, dynamic>{
'field': instance.field,
'direction': instance.direction,
packages/stream_chat/lib/src/api/responses.dart
+7 -8
View File
@@ -1,15 +1,14 @@
import 'package:json_annotation/json_annotation.dart';
import 'package:stream_chat/src/client.dart';
import 'package:stream_chat/src/models/channel_model.dart';
import 'package:stream_chat/src/models/channel_state.dart';
import 'package:stream_chat/src/models/device.dart';
import 'package:stream_chat/src/models/event.dart';
import '../models/channel_model.dart';
import '../models/channel_state.dart';
import '../models/member.dart';
import '../models/message.dart';
import '../models/reaction.dart';
import '../models/read.dart';
import '../models/user.dart';
import 'package:stream_chat/src/models/member.dart';
import 'package:stream_chat/src/models/message.dart';
import 'package:stream_chat/src/models/reaction.dart';
import 'package:stream_chat/src/models/read.dart';
import 'package:stream_chat/src/models/user.dart';
part 'responses.g.dart';
packages/stream_chat/lib/src/api/retry_policy.dart
+2 -1
View File
@@ -18,7 +18,8 @@ class RetryPolicy {
final bool Function(StreamChatClient client, int attempt, ApiError apiError)
shouldRetry;
/// In the case that we want to retry a failed request the retryTimeout method is called to determine the timeout
/// In the case that we want to retry a failed request the retryTimeout
/// method is called to determine the timeout
final Duration Function(
StreamChatClient client, int attempt, ApiError apiError) retryTimeout;
packages/stream_chat/lib/src/api/retry_queue.dart
+7 -7
View File
@@ -1,8 +1,8 @@
import 'dart:async';
import 'package:collection/collection.dart';
import 'package:meta/meta.dart';
import 'package:logging/logging.dart';
import 'package:meta/meta.dart';
import 'package:stream_chat/src/api/channel.dart';
import 'package:stream_chat/src/api/retry_policy.dart';
import 'package:stream_chat/src/event_type.dart';
@@ -12,12 +12,6 @@ import 'package:stream_chat/stream_chat.dart';
/// The retry queue associated to a channel
class RetryQueue {
/// The channel of this queue
final Channel channel;
/// The logger associated to this queue
final Logger logger;
/// Instantiate a new RetryQueue object
RetryQueue({
@required this.channel,
@@ -30,6 +24,12 @@ class RetryQueue {
_listenFailedEvents();
}
/// The channel of this queue
final Channel channel;
/// The logger associated to this queue
final Logger logger;
final _subscriptions = <StreamSubscription>[];
void _listenConnectionRecovered() {
packages/stream_chat/lib/src/api/websocket.dart
+39 -38
View File
@@ -2,28 +2,29 @@ import 'dart:async';
import 'dart:convert';
import 'dart:math';
import 'package:meta/meta.dart';
import 'package:logging/logging.dart';
import 'package:meta/meta.dart';
import 'package:rxdart/rxdart.dart';
import 'package:web_socket_channel/web_socket_channel.dart';
import '../models/event.dart';
import '../models/user.dart';
import 'connection_status.dart';
import 'web_socket_channel_stub.dart'
import 'package:stream_chat/src/api/connection_status.dart';
import 'package:stream_chat/src/api/web_socket_channel_stub.dart'
if (dart.library.html) 'web_socket_channel_html.dart'
if (dart.library.io) 'web_socket_channel_io.dart';
import 'package:stream_chat/src/models/event.dart';
import 'package:stream_chat/src/models/user.dart';
import 'package:web_socket_channel/web_socket_channel.dart';
/// Typedef which exposes an [Event] as the only parameter.
typedef EventHandler = void Function(Event);
/// Typedef used for connecting to a websocket. Method returns a [WebSocketChannel]
/// and accepts a connection [url] and an optional [Iterable] of `protocols`.
/// Typedef used for connecting to a websocket. Method returns a
/// [WebSocketChannel] and accepts a connection [url] and an optional
/// [Iterable] of `protocols`.
typedef ConnectWebSocket = WebSocketChannel Function(String url,
{Iterable<String> protocols});
// TODO: parse error even
// TODO: if parsing an error into an event fails we should not hide the original error
// TODO: if parsing an error into an event fails we should not hide the
// TODO: original error
/// A WebSocket connection that reconnects upon failure.
class WebSocket {
/// Creates a new websocket
@@ -75,7 +76,8 @@ class WebSocket {
/// WS connection payload
final Map<String, dynamic> connectPayload;
/// Functions that will be called every time a new event is received from the connection
/// Functions that will be called every time a new event is received from the
/// connection
final EventHandler handler;
/// A WS specific logger instance
@@ -87,8 +89,9 @@ class WebSocket {
final ConnectWebSocket connectFunc;
/// Interval of the reconnection monitor timer
/// This checks that it received a new event in the last [reconnectionMonitorTimeout] seconds,
/// otherwise it considers the connection unhealthy and reconnects the WS
/// This checks that it received a new event in the last
/// [reconnectionMonitorTimeout] seconds, otherwise it considers the
/// connection unhealthy and reconnects the WS
final int reconnectionMonitorInterval;
/// Interval of the health event sending timer
@@ -96,7 +99,8 @@ class WebSocket {
/// make the server aware that the client is still listening
final int healthCheckInterval;
/// The timeout that uses the reconnection monitor timer to consider the connection unhealthy
/// The timeout that uses the reconnection monitor timer to consider the
/// connection unhealthy
final int reconnectionMonitorTimeout;
final _connectionStatusController =
@@ -121,9 +125,7 @@ class WebSocket {
_connecting = false,
_reconnecting = false;
Event _decodeEvent(String source) {
return Event.fromJson(json.decode(source));
}
Event _decodeEvent(String source) => Event.fromJson(json.decode(source));
Completer<Event> _connectionCompleter = Completer<Event>();
@@ -166,8 +168,8 @@ class WebSocket {
return;
}
logger.info(
'connection closed | closeCode: ${_channel.closeCode} | closedReason: ${_channel.closeReason}');
logger.info('connection closed | closeCode: ${_channel.closeCode} | '
'closedReason: ${_channel.closeReason}');
if (!_reconnecting) {
_reconnect();
@@ -200,8 +202,7 @@ class WebSocket {
}
Future<void> _onConnectionError(error, [stacktrace]) async {
logger.severe('error connecting');
logger.severe(error);
logger..severe('error connecting')..severe(error);
if (stacktrace != null) {
logger.severe(stacktrace);
}
@@ -219,21 +220,21 @@ class WebSocket {
}
}
void _startReconnectionMonitor() {
final reconnectionTimer = (_) {
final now = DateTime.now();
if (_lastEventAt != null &&
now.difference(_lastEventAt).inSeconds > reconnectionMonitorTimeout) {
_channel.sink.close();
}
};
void _reconnectionTimer(_) {
final now = DateTime.now();
if (_lastEventAt != null &&
now.difference(_lastEventAt).inSeconds > reconnectionMonitorTimeout) {
_channel.sink.close();
}
}
void _startReconnectionMonitor() {
_reconnectionMonitor = Timer.periodic(
Duration(seconds: reconnectionMonitorInterval),
reconnectionTimer,
_reconnectionTimer,
);
reconnectionTimer(_reconnectionMonitor);
_reconnectionTimer(_reconnectionMonitor);
}
void _reconnectTimer() async {
@@ -283,20 +284,20 @@ class WebSocket {
}
}
void _healthCheckTimer(_) {
logger.info('sending health.check');
_channel.sink.add("{'type': 'health.check'}");
}
void _startHealthCheck() {
logger.info('start health check monitor');
final healthCheckTimer = (_) {
logger.info('sending health.check');
_channel.sink.add("{'type': 'health.check'}");
};
_healthCheck = Timer.periodic(
Duration(seconds: healthCheckInterval),
healthCheckTimer,
_healthCheckTimer,
);
healthCheckTimer(_healthCheck);
_healthCheckTimer(_healthCheck);
}
/// Disconnects the WS and releases eventual resources