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

Add custom sort support for offline channels

Browse Source 
Signed-off-by: Sahil Kumar <[email protected]>
This commit is contained in:
Sahil Kumar
2021-02-26 18:30:22 +05:30
parent aeecc71154
commit 623d48e32b
13 changed files with 394 additions and 442 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/stream_chat/lib/src/client.dart
+113 -140
View File
@@ -9,6 +9,7 @@ import 'package:rxdart/rxdart.dart';
import 'package:stream_chat/src/api/retry_policy.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_model.dart';
import 'package:stream_chat/src/models/own_user.dart';
import 'package:stream_chat/src/platform_detector/platform_detector.dart';
import 'package:stream_chat/version.dart';
@@ -22,6 +23,7 @@ import 'api/responses.dart';
import 'api/websocket.dart';
import 'db/chat_persistence_client.dart';
import 'exceptions.dart';
import 'models/channel_state.dart';
import 'models/event.dart';
import 'models/message.dart';
import 'models/user.dart';
@@ -489,7 +491,7 @@ class StreamChatClient {
if (status == ConnectionStatus.connected &&
state.channels?.isNotEmpty == true) {
unawaited(queryChannels(filter: {
unawaited(_queryChannels(filter: {
'cid': {
'\$in': state.channels.keys.toList(),
},
@@ -578,15 +580,15 @@ class StreamChatClient {
final _queryChannelsStreams = <String, Future<List<Channel>>>{};
/// Requests channels with a given query.
Future<List<Channel>> queryChannels({
Stream<List<Channel>> queryChannels({
Map<String, dynamic> filter,
List<SortOption> sort,
List<SortOption<ChannelModel>> sort,
Map<String, dynamic> options,
PaginationParams paginationParams = const PaginationParams(limit: 10),
int messageLimit,
bool onlyOffline = false,
bool preferOffline = true,
bool waitForConnect = true,
}) async {
}) async* {
if (waitForConnect) {
if (_connectCompleter != null && !_connectCompleter.isCompleted) {
logger.info('awaiting connection completer');
@@ -598,7 +600,7 @@ class StreamChatClient {
if (persistenceEnabled) {
logger.warning(
'$errorMessage\nTrying to retrieve channels from the offline storage.');
onlyOffline = true;
preferOffline = true;
} else {
throw Exception(errorMessage);
}
@@ -606,34 +608,41 @@ class StreamChatClient {
}
final hash = base64.encode(utf8.encode(
'$filter${_asMap(sort)}$options${paginationParams?.toJson()}$messageLimit$onlyOffline'));
'$filter${_asMap(sort)}$options${paginationParams?.toJson()}$messageLimit$preferOffline',
));
if (_queryChannelsStreams.containsKey(hash)) {
return _queryChannelsStreams[hash];
yield await _queryChannelsStreams[hash];
} else {
if (true) {
final channels = await _queryChannelsOffline(
filter: filter,
sort: sort,
paginationParams: paginationParams,
);
if (channels.isNotEmpty) yield channels;
}
final newQueryChannelsFuture = _queryChannels(
filter: filter,
sort: sort,
options: options,
paginationParams: paginationParams,
messageLimit: messageLimit,
);
_queryChannelsStreams[hash] = newQueryChannelsFuture;
yield await newQueryChannelsFuture;
}
final newQueryChannelsStream = _doQueryChannels(
filter: filter,
sort: sort,
options: options,
paginationParams: paginationParams,
messageLimit: messageLimit,
onlyOffline: onlyOffline,
).whenComplete(() {
_queryChannelsStreams.remove(hash);
});
_queryChannelsStreams[hash] = newQueryChannelsStream;
return newQueryChannelsStream;
}
Future<List<Channel>> _doQueryChannels({
Future<List<Channel>> _queryChannels({
@required Map<String, dynamic> filter,
@required List<SortOption> sort,
@required Map<String, dynamic> options,
@required int messageLimit,
List<SortOption<ChannelModel>> sort,
Map<String, dynamic> options,
int messageLimit,
PaginationParams paginationParams = const PaginationParams(limit: 10),
bool onlyOffline = false,
}) async {
logger.info('Query channel start');
final defaultOptions = {
@@ -661,86 +670,85 @@ class StreamChatClient {
payload.addAll(paginationParams.toJson());
}
if (onlyOffline) {
return _queryChannelsOffline(
filter: filter,
sort: sort,
paginationParams: paginationParams,
);
}
final response = await get(
'/channels',
queryParameters: {
'payload': jsonEncode(payload),
},
);
try {
final response = await get(
'/channels',
queryParameters: {
'payload': jsonEncode(payload),
},
);
final res = decode<QueryChannelsResponse>(
response.data,
QueryChannelsResponse.fromJson,
);
final res = decode<QueryChannelsResponse>(
response.data,
QueryChannelsResponse.fromJson,
);
final users = res.channels
?.expand((channel) => channel.members.map((member) => member.user))
?.toList();
if (users != null) {
state._updateUsers(users);
}
logger.info('Got ${res.channels?.length} channels from api');
if (res.channels?.isEmpty != false &&
(paginationParams?.offset ?? 0) == 0) {
logger.warning('''We could not find any channel for this query.
if (res.channels?.isEmpty == true && (paginationParams?.offset ?? 0) == 0) {
logger.warning('''We could not find any channel for this query.
Please make sure to take a look at the Flutter tutorial: https://getstream.io/chat/flutter/tutorial
If your application already has users and channels, you might need to adjust your query channel as explained in the docs https://getstream.io/chat/docs/query_channels/?language=dart''');
}
final newChannels = Map<String, Channel>.from(state.channels ?? {});
final channels = <Channel>[];
if (res.channels != null) {
for (final channelState in res.channels) {
final channel = newChannels[channelState.channel.cid];
if (channel != null) {
channel.state?.updateChannelState(channelState);
channels.add(channel);
} else {
final newChannel = Channel.fromState(this, channelState);
await chatPersistenceClient
?.updateChannelState(newChannel.state.channelState);
newChannel.state?.updateChannelState(channelState);
newChannels[newChannel.cid] = newChannel;
channels.add(newChannel);
}
}
}
state.channels = newChannels;
await chatPersistenceClient?.updateChannelQueries(
filter,
res.channels.map((c) => c.channel.cid).toList(),
paginationParams?.offset == null || paginationParams.offset == 0,
);
return channels;
} catch (e) {
if (!persistenceEnabled) {
rethrow;
}
return _queryChannelsOffline(
filter: filter,
sort: sort,
paginationParams: paginationParams,
);
return <Channel>[];
}
final channels = res.channels;
final users = channels
.expand((it) => it.members)
.map((it) => it.user)
.toList(growable: false);
state._updateUsers(users);
logger.info('Got ${res.channels?.length} channels from api');
final updateData = _mapChannelStateToChannel(channels);
await chatPersistenceClient?.updateChannelQueries(
filter,
channels.map((c) => c.channel.cid).toList(),
paginationParams?.offset == null || paginationParams.offset == 0,
);
state.channels = updateData.key;
return updateData.value;
}
dynamic _parseError(DioError error) {
Future<List<Channel>> _queryChannelsOffline({
@required Map<String, dynamic> filter,
@required List<SortOption<ChannelModel>> sort,
PaginationParams paginationParams = const PaginationParams(limit: 10),
}) async {
final offlineChannels = await chatPersistenceClient?.getChannelStates(
filter: filter,
sort: sort,
paginationParams: paginationParams,
);
final updatedData = _mapChannelStateToChannel(offlineChannels);
state.channels = updatedData.key;
return updatedData.value;
}
MapEntry<Map<String, Channel>, List<Channel>> _mapChannelStateToChannel(
List<ChannelState> channelStates,
) {
final channels = {...state.channels ?? {}};
final newChannels = <Channel>[];
if (channelStates != null) {
for (final channelState in channelStates) {
final channel = channels[channelState.channel.cid];
if (channel != null) {
channel.state?.updateChannelState(channelState);
newChannels.add(channel);
} else {
final newChannel = Channel.fromState(this, channelState);
channels[newChannel.cid] = newChannel;
newChannels.add(newChannel);
}
}
}
return MapEntry(channels, newChannels);
}
Object _parseError(DioError error) {
if (error.type == DioErrorType.RESPONSE) {
final apiError =
ApiError(error.response?.data, error.response?.statusCode);
@@ -751,39 +759,6 @@ class StreamChatClient {
return error;
}
Future<List<Channel>> _queryChannelsOffline({
@required Map<String, dynamic> filter,
@required List<SortOption> sort,
PaginationParams paginationParams = const PaginationParams(limit: 10),
}) async {
final offlineChannels = await chatPersistenceClient?.getChannelStates(
filter: filter,
sort: sort,
paginationParams: paginationParams,
) ??
[];
final newChannels = Map<String, Channel>.from(state.channels ?? {});
logger.info('Got ${offlineChannels.length} channels from storage');
final channels = offlineChannels.map((channelState) {
final channel = newChannels[channelState.channel.cid];
if (channel != null) {
channel.state?.updateChannelState(channelState);
return channel;
} else {
final newChannel = Channel.fromState(this, channelState);
chatPersistenceClient
?.updateChannelState(newChannel.state.channelState);
newChannels[newChannel.cid] = newChannel;
return newChannel;
}
}).toList();
if (channels.isNotEmpty) {
state.channels = newChannels;
}
return channels;
}
/// Handy method to make http GET request with error parsing.
Future<Response<String>> get(
String path, {
@@ -1448,18 +1423,16 @@ class ClientState {
_userController.add(user);
}
void _updateUsers(List<User> users) {
users?.forEach(_updateUser);
}
void _updateUser(User user) {
void _updateUsers(List<User> userList) {
final newUsers = {
...users ?? {},
user.id: user,
for (var user in userList) user.id: user,
};
_usersController.add(newUsers);
}
void _updateUser(User user) => _updateUsers([user]);
/// The current user
OwnUser get user => _userController.value;