flutter getx web sockets -流数据级联到提供商和控制器

Question

目标很简单

• flutter app通过websockets调用graphql api
• 应用视图调用控制器，控制器调用提供商，提供商通过websockets或HTTP api socket调用AWS appsync api
• 我们从appsync api或HTTP api socket调用的数据流通过websockets每隔一段时间从后端
• 流需要级联回提供商，然后到控制器(这是关键的一步)
• 控制器(不是提供者)将更新对象或响应变量，使UI反映更改

问题:数据是通过调用者的websockets接收的，但从未作为流传递给提供者或控制器以反映更改

示例代码实际调用者orderdata.dart

@override
Stream<dynamic> subscribe({
String query,
Map<String, dynamic> variables,
}) async* {
debugPrint('===->subscribe===');
// it can be any stream here, http or file or image or media
final Stream<GraphQLResponse<String>> operation = Amplify.API.subscribe(
GraphQLRequest<String>(
document: query,
variables: variables,
),
onEstablished: () {
debugPrint(
'===->subscribe onEstablished ===',
);
},
);
operation.listen(
(event) async* {
final jsonData = json.decode(event.data.toString());
debugPrint('===->subscription data $jsonData');
yield jsonData;
},
onError: (Object e) => debugPrint('Error in subscription stream: $e'),
);
}

提供程序中的orderprovider.dart

Stream<Order> orderSubscription(String placeId) async* {
debugPrint('===->=== $placeId');
subscriptionResponseStream = orderData.subscribe(
query: subscribeToMenuOrder,
variables: {"place_id": placeId},
);
subscriptionResponseStream.listen((event) async* {
debugPrint(
"===->=== yielded $event",
);
yield event;
});
debugPrint('===->=== finished');
}

控制器中的homecontroller.dart

Future<void> getSubscriptionData(String placeId) async {
debugPrint('===HomeController->getSubscriptionData===');
OrderProvider().orderSubscription(placeId).listen(
(data) {
//this block is executed when data event is receivedby listener
debugPrint('Data: $data');
Get.snackbar('orderSubscription', data.toString());
},
onError: (err) {
//this block is executed when error event is received by listener
debugPrint('Error: $err');
},
cancelOnError:
false, //this decides if subscription is cancelled on error or not
onDone: () {
//this block is executed when done event is received by listener
debugPrint('Done!');
},
);
}

homeview调用homecontroller

Answer 1

尝试使用map转换流:

@override
Stream<dynamic> subscribe({
String query,
Map<String, dynamic> variables,
}) {
debugPrint('===->subscribe===');
// it can be any stream here, http or file or image or media
final Stream<GraphQLResponse<String>> operation = Amplify.API.subscribe(
GraphQLRequest<String>(
document: query,
variables: variables,
),
onEstablished: () {
debugPrint(
'===->subscribe onEstablished ===',
);
},
);

return operation.map((event) {
return json.decode(event.data);
});
}
// elsewhere
final subscription = subscribe(
query: 'some query', 
variables: {},
);
subscription.listen(
(jsonData) {
debugPrint('===->subscription data $jsonData');
},
onError: (Object e) => debugPrint('Error in subscription stream: $e'),
);