OfflineMutationTypedLink - How to pass same client to OfflineMutationTypedLink

[francescreig]

Hello,

I am been working on creating an OfflineMutationTypedLink same as this PR #147. I've nearly implemented but I have a problem that actually it was pointed here: when you want to reexecute offline mutations, calling requestController.add(req) would not execute the request since it does not see any response.
Trying to instead call client.request(req) (see below)

void _handleOnConnect() => mutationQueueBox.values.forEach((json) async {
      final OperationRequest req = serializers.deserialize(json);

      // Run unexecuted mutations
      await client.request(req).firstWhere((res) => res.dataSource == DataSource.Link); <---- here
 });

leads me to call again the request method that I override on the same OfflineMutationTypedLink, so I end up in a recursive call that freezes the app.

Here it is pointed to pass the client as a dependency, but I think I am failing on this, since using the same client (OfflineClient) that internally has this OfflineMutationTypedLink gives a recursive calls.

This is my request method that I override:

@override
Stream<OperationResponse<TData, TVars>> request<TData, TVars>(
  OperationRequest<TData, TVars> request, [
  NextTypedLink<TData, TVars>? forward,
]) {
  return _handleRequest(request, forward).transform(_responseTransformer());
}

Stream<OperationResponse<TData, TVars>> _handleRequest<TData, TVars>(
  OperationRequest<TData, TVars> request, [
  NextTypedLink<TData, TVars>? forward,
]) {
  // Forward along any operations that aren't Mutations
  if (_isMutation(request) == false) return forward!(request);

  if (connectionStatus.hasConnection) {
    connected = true;
  } else {
    connected = false;
  }
  if (connected) return forward!(request);

  // Save mutation to the queue
  mutationQueueBox.add(serializers.serialize(request));

  // Add an optimistic patch to the cache, if necessary
  if (request.optimisticResponse != null) {
    cache.writeQuery(
      request,
      request.optimisticResponse,
      optimisticRequest: config?.persistOptimisticResponse == false ? request : null,
    );
    return forward!(request);
  }

  // Don't forward
  return NeverStream();
}

So when I set the setter connected to true it runs a method to execute the requests, thus entering in a recursive call (calling request overrided method from above):

void _handleOnConnect() async {
  for (final json in mutationQueueBox.values) {
    final req = serializers.deserialize(json);

    // Run unexecuted mutations
    await client
        .request(req)
        .firstWhere((res) => res.dataSource == DataSource.Link && res.operationRequest == req);
  }
}

[knaeckeKami]

You'd need to pass in the actual client into the offline mutation Link, so it can call the real request () method.

[francescreig]

In my OfflineMutationTypedLink I have this contructor:

OfflineMutationTypedLink({
  required this.mutationQueueBox,
  required this.serializers,
  required this.cache,
  required this.requestController,
  // required this.client, <----- do I need to pass the client here? (client is OfflineClient type)
  this.config = const OfflineClientConfig(),
});

If I pass it here, when I am declaring the OfflineClient, has to be declared like this?

OfflineClient({
  required this.link,
  required this.storeBox,
  required this.mutationQueueBox,
  required this.serializers,
  this.offlineConfig,
  required this.requestController,
  this.typePolicies = const {},
  this.updateCacheHandlers = const {},
  this.defaultFetchPolicies = const {},
  this.addTypename = true,
}) : cache = Cache(
        store: HiveStore(storeBox),
        typePolicies: typePolicies,
        addTypename: addTypename,
      ) {
  _typedLink = TypedLink.from([
    RequestControllerTypedLink(requestController),
    OfflineMutationTypedLink(
      cache: cache,
      mutationQueueBox: mutationQueueBox,
      serializers: serializers,
      requestController: requestController,
      config: offlineConfig,
      // client: this <----- here adding this also enters in recursive
    ),
    if (addTypename) const AddTypenameTypedLink(),
    if (updateCacheHandlers.isNotEmpty)
      UpdateCacheTypedLink(
        cache: cache,
        updateCacheHandlers: updateCacheHandlers,
      ),
    FetchPolicyTypedLink(
      link: link,
      cache: cache,
      defaultFetchPolicies: defaultFetchPolicies,
    )
  ]);
}

Thanks

[knaeckeKami]

There is a chicken - and egg problem here, since the link needs the client and the client needs the link.

You could first create the link, then the client, and then set the Client in the link.

[francescreig]

Okay, thank you @knaeckeKami ! Finally I also discovered that my code when I had connection I was setting always connect to true and thus it was always calling request making the infinite recursive loop. Adding instead a listener into the request handler solved the issue.
A part from that, now I am struggling with a type problem that it was also pointed on the same discussion (here).
So trying this:

void _handleOnConnect() async {
  final queue = StreamQueue(Stream.fromIterable(mutationQueueBox.values));
  while (await queue.hasNext) {
    final json = await queue.next;
    final req = serializers.deserialize(json) as OperationRequest<Object, Object>;
    // Run unexecuted mutations
    await client
        .request(req)
        .firstWhere((res) => res.dataSource == DataSource.Link && res.operationRequest == req);
  }
}

results to an Unhandled Exception on the UpdateChaceTypedLink, since it does not match the expected Type:

[VERBOSE-2:dart_vm_initializer.cc(41)] Unhandled Exception: type '(CacheProxy, OperationResponse<GUpsertObservationData, GUpsertObservationVars>) => void' is not a subtype of type '(CacheProxy, OperationResponse<Object, Object>) => void' in type cast
#0      UpdateCacheTypedLink._updateCache
#1      _DoStreamSink.onData
do.dart:31
#2      _RootZone.runUnaryGuarded (dart:async/zone.dart:1593:10)
#3      _BufferingStreamSubscription._sendData (dart:async/stream_impl.dart:339:11)
#4      _BufferingStreamSubscription._add (dart:async/stream_impl.dart:271:7)
#5      _SyncStreamControllerDispatch._sendData (dart:async/stream_controller.dart:774:19)
#6      _StreamController._add (dart:async/stream_controller.dart:648:7)
#7      _StreamController.add (dart:async/stream_controller.dart:596:5)
#8      _DoStreamSink.onData
do.dart:40
#9      _RootZone.runUnaryGuarded (dart:async/zone.dart:1593:10)
#10     _BufferingStreamSubscription._sendData (dart:async/stream_impl.dart:339:11)
#11     _BufferingStreamSubscription._add (dart:async/stream_impl.dart:271:7)
#12     _SyncStreamControllerDispatch._sendData (dart:async/stream_controller.dart:774:19)
#13     _StreamController._add (dart:async/stream_controller.dart:648:7)
#14     _StreamController.add (dart:async/stream_controller.dart:596:5)
#15     _DoStreamSink.onData
do.dart:40
#16     _RootZone.runUnaryGuarded (dart:async/zone.dart:1593:10)

I could infere the type depending on the mutation I am trying to deserialize, but it is possible that dart could assign the correct type?

Thank you beforehand!

[knaeckeKami]

unfortunately, that's an issue with variance here.

I don't have a good solution at the moment, but I think a workaround would be changing your cachehandlers to accept OperationResponse<Object?, Object?> and cast the results

[francescreig]

Okay thank you. For the moment defining a cache handler type as UpdateCacheHandler<Object?, Object?> and type casting inside the cache handler like for instance:

final responseUpsertObservation = response as OperationResponse<GUpsertObservationData, GUpsertObservationVars>;

Moves the error from the OfflineMutationTypedLink to the cache handler, but still cannot cast the subtype giving the same typecasting error:

Unhandled Exception: type 'OperationResponse<Object, Object>' is not a subtype of type 'OperationResponse<GUpsertObservationData, GUpsertObservationVars>' in type cast

I will try to investigate if it is possible to cast it somehow. Do you know if there's probably an internal issue (or using built_value underhood library?)

[knaeckeKami]

You can work around it by casting not the response, but the data:

void myCacheHandler(
  CacheProxy proxy,
  OperationResponse<dynamic, dynamic> response,
) {
  //ignore: avoid_dynamic_calls
  final myData = response.data as MyData?

}

Using dynamic instead of Object? is usually a bad idea, but in this instance it will get Dart to shut up ;)

[knaeckeKami]

I will try to investigate if it is possible to cast it somehow. Do you know if there's probably an internal issue (or using built_value underhood library?)

This is not an issue in built_value. Actually, built_value is one of the only serialization libraries that support something like

serializers.deserialize(json);

which would deserialize to the correct type without you having to specify it. (Most would require you to write MyReq.fromJson(json) but would not allow you to just deserialize any request without specifying which one ).

The issue is that after deserialization, we don't know the exact type. We just know it an OperationRequest<Object, Object>, so we can cast it up to this type.
Which works fine in covariant positions (an OperationResponse<MyData, MyVars> is an OperationResponse<Object, Object>) for return types, but not in contravariant positions (a void Function(OperationResponse<MyData, MyVars>) is NOT a void Function(OperationResponse<Object, Object>),

So, even if all the runtime types match perfectly, Dart still checks the types before calling the function and because the generics type parameters are <Object, Object> instead of <MyData, MyVars>, it will throw an exception.

This is unfortunately more of an ugly corner of the Dart type system than an issue in built_value.

However, there would be ways around this.

We could add a new method to OperationRequest:

abstract class OperationRequest<TData, TVars> {

   Stream<OperationResponse<TData, TVars> execute(TypedLink link) => link.request(this);
}

With this hack, we could get Dart to infer the correct type parameters.

This would require changes to the code generation though since OperationRequest is implemented, not extended at the moment.

[francescreig]

Hmmm interesting... 🤔 Thank you for the clarification! I finally could get rid of the type error as you suggested casting the response.data on my handler.
I was thinking that probably since I got the hands dirty with the OfflineMutationTypedLink try to push the PR #144 and close it since I am also trying to implement what @akinsho did in the past.

[akinsho]

@francescreig please feel free to carry on with that PR, I've since moved project and role so am no longer able to push that PR forward. It's been quite a while since I looked at any of that code so not sure I can remember enough to provide any useful input

[knaeckeKami]

Hmmm interesting... 🤔 Thank you for the clarification! I finally could get rid of the type error as you suggested casting the response.data on my handler. I was thinking that probably since I got the hands dirty with the OfflineMutationTypedLink try to push the PR #144 and close it since I am also trying to implement what @akinsho did in the past.

Yeah, I was also running into this issue myself and did not really understand it, until the same thing happened when it added the feature to run the client in an Isolate (also there the requests are serialized and the receiver does not know the. generic types), and so I had to do something similar ->
create a command with the correct generic types, send the command over the isolate, and let the command call the request() method itself without the receiver needing to infer the types:

ferry/packages/ferry/lib/src/isolate/isolate_commands.dart

Line 72 in f253e43

class RequestCommand<TData, TVars> extends IsolateCommand {

If you manage to improve the OfflineMutationTypedLink, I'd be grateful. As far as I understand, the currently shipped OfflineMutationLink does not work at all, right? (I was not the maintainer of ferry at the time this was written and did not use it myself yet so I may not fully understand what is happening there)

[vorte]

@francescreig can you share your solution? Or if possible add this to the open PR? Either would be very helpful.