Streaming support in React Native

[smaye81]

Currently, Connect Web cannot support server-streaming via the Connect protocol in React Native. This is due to various limitations in React Native's implementation of the Fetch API. This issue is to document the state of things as well as potentially track any effort on our side that would be needed if/when the below shortcomings are resolved.

In a nutshell, React Native uses Github's fetch polyfill to provide a subset of the Fetch API. However, this polyfill is built upon XMLHttpRequest, which doesn't offer full streaming natively (doc) so the polyfill does not support streaming requests. Additionally, there are no plans to fix this. So the current state of React Native as-is isn't a viable option. For additional context, see the following issues:

• fetch implementation does not support streams from the spec - needs implementation in React Native core facebook/react-native#27741
• Implement response.body as a stream JakeChampion/fetch#746

Another polyfill exists as a fork of this called react-native-fetch-api, which is built on top of React Native's networking API instead of XMLHttpRequest and aims to 'fill in some gaps of the WHATWG specification for fetch, namely the support for text streaming.'

But, React Native's Networking API also does not support full streaming natively. The fork provides text streaming only (see the README for an explanation why). Connect streaming RPCs are binary, though. Even with the JSON format, the enveloping is binary. So, this approach is not viable either. The fork's README mentions that if you wish to consume binary data, 'either blob or base64 response types have to be used', but Connect does not implement the gRPC-Web-Text (base64 encoded) format, so again, not an option.

The natural thought is 'ok, can't you just write your own polyfill that fixes all these problems so that Connect streaming works', but that isn't feasible either. Perhaps the best explanation is from this comment on the Github fetch polyfill's streaming issue linked above. Basically:

If you want actual streaming support in React Native, then React Native would need to implement it itself.

Taking all this into account, there is no real option to implement full streaming on React Native for Connect at the moment. Hopefully this changes in the future, but for right now, there is unfortunately nothing we easily do to remedy it.

[Madumo]

I'll put this here since this might help someone like me that need to use the binary protocol within a react native app. The provided transports uses ReadableStream, which is not yet available in React Native. I tried all the ways i could find to polyfill it, but nothing worked.

I ended up implementing a custom transport that does unary requests using the grpc-web protocol in binary format. To do this, I use xhr directly which allow us to set the response type to arraybuffer and let us get the response in a binary format, and then parse it. With this, we lose the optimisation of parsing data as it arrives while the request is in progress, but at least it works with React Native.

import { Message } from '@bufbuild/protobuf';
import type {
  AnyMessage,
  MethodInfo,
  PartialMessage,
  ServiceType,
} from '@bufbuild/protobuf';
import type { UnaryRequest } from '@bufbuild/connect';
import { Code, connectErrorFromReason, runUnary } from '@bufbuild/connect';
import type {
  StreamResponse,
  Transport,
  UnaryResponse,
} from '@bufbuild/connect';
import {
  createClientMethodSerializers,
  createMethodUrl,
  encodeEnvelope,
} from '@bufbuild/connect/protocol';
import {
  requestHeader,
  trailerFlag,
  trailerParse,
  validateResponse,
  validateTrailer,
} from '@bufbuild/connect/protocol-grpc-web';
import { GrpcWebTransportOptions } from '@bufbuild/connect-web';

class AbortError extends Error {
  name = 'AbortError';
}

interface FetchXHRResponse {
  status: number;
  headers: Headers;
  body: Uint8Array;
}

function parseHeaders(allHeaders: string): Headers {
  return allHeaders
    .trim()
    .split(/[\r\n]+/)
    .reduce((memo, header) => {
      const [key, value] = header.split(': ');
      memo.append(key, value);
      return memo;
    }, new Headers());
}

function extractDataChunks(initialData: Uint8Array) {
  let buffer = initialData;
  let dataChunks: { flags: number; data: Uint8Array }[] = [];

  while (buffer.byteLength >= 5) {
    let length = 0;
    let flags = buffer[0];

    for (let i = 1; i < 5; i++) {
      length = (length << 8) + buffer[i]; // eslint-disable-line no-bitwise
    }

    const data = buffer.subarray(5, 5 + length);
    buffer = buffer.subarray(5 + length);
    dataChunks.push({ flags, data });
  }

  return dataChunks;
}

export function createXHRGrpcWebTransport(
  options: GrpcWebTransportOptions
): Transport {
  const useBinaryFormat = options.useBinaryFormat ?? true;
  return {
    async unary<
      I extends Message<I> = AnyMessage,
      O extends Message<O> = AnyMessage
    >(
      service: ServiceType,
      method: MethodInfo<I, O>,
      signal: AbortSignal | undefined,
      timeoutMs: number | undefined,
      header: Headers,
      message: PartialMessage<I>
    ): Promise<UnaryResponse<I, O>> {
      const { normalize, serialize, parse } = createClientMethodSerializers(
        method,
        useBinaryFormat,
        options.jsonOptions,
        options.binaryOptions
      );

      try {
        return await runUnary<I, O>(
          {
            stream: false,
            service,
            method,
            url: createMethodUrl(options.baseUrl, service, method),
            init: {
              method: 'POST',
              mode: 'cors',
            },
            header: requestHeader(useBinaryFormat, timeoutMs, header),
            message: normalize(message),
            signal: signal ?? new AbortController().signal,
          },
          async (req: UnaryRequest<I, O>): Promise<UnaryResponse<I, O>> => {
            function fetchXHR(): Promise<FetchXHRResponse> {
              return new Promise((resolve, reject) => {
                const xhr = new XMLHttpRequest();

                xhr.open(req.init.method ?? 'POST', req.url);

                function onAbort() {
                  xhr.abort();
                }

                req.signal.addEventListener('abort', onAbort);

                xhr.addEventListener('abort', () => {
                  reject(new AbortError('Request aborted'));
                });

                xhr.addEventListener('load', () => {
                  resolve({
                    status: xhr.status,
                    headers: parseHeaders(xhr.getAllResponseHeaders()),
                    body: new Uint8Array(xhr.response),
                  });
                });

                xhr.addEventListener('error', () => {
                  reject(new Error('Network Error'));
                });

                xhr.addEventListener('loadend', () => {
                  req.signal.removeEventListener('abort', onAbort);
                });

                xhr.responseType = 'arraybuffer';

                req.header.forEach((value: string, key: string) =>
                  xhr.setRequestHeader(key, value)
                );

                xhr.send(encodeEnvelope(0, serialize(req.message)));
              });
            }

            const response = await fetchXHR();

            validateResponse(
              useBinaryFormat,
              response.status,
              response.headers
            );

            const chunks = extractDataChunks(response.body);

            let trailer: Headers | undefined;
            let message: O | undefined;

            chunks.forEach(({ flags, data }) => {
              if (flags === trailerFlag) {
                if (trailer !== undefined) {
                  throw 'extra trailer';
                }

                // Unary responses require exactly one response message, but in
                // case of an error, it is perfectly valid to have a response body
                // that only contains error trailers.
                trailer = trailerParse(data);
                return;
              }

              if (message !== undefined) {
                throw 'extra message';
              }

              message = parse(data);
            });

            if (trailer === undefined) {
              throw 'missing trailer';
            }

            validateTrailer(trailer);

            if (message === undefined) {
              throw 'missing message';
            }

            return <UnaryResponse<I, O>>{
              stream: false,
              header: response.headers,
              message,
              trailer,
            };
          },
          options.interceptors
        );
      } catch (e) {
        throw connectErrorFromReason(e, Code.Internal);
      }
    },

    async stream<
      I extends Message<I> = AnyMessage,
      O extends Message<O> = AnyMessage
    >(
      _service: ServiceType,
      _method: MethodInfo<I, O>,
      _signal: AbortSignal | undefined,
      _timeoutMs: number | undefined,
      _header: HeadersInit_ | undefined,
      _input: AsyncIterable<I>
    ): Promise<StreamResponse<I, O>> {
      return Promise.reject(Error('NOT IMPLEMENTED'));
    },
  };
}

[hollanderbart]

@smaye81 When do you expect to have server-streaming support with Connect in React Native? We're planning to use it in our app to always get the latest state of IOT devices. Thanks!

[smaye81]

Hi @hollanderbart. In fairness, this issue is somewhat misleading. There is currently no option to provide server-streaming support in React Native due to some limitations with the Fetch API. We would need action on React Native's part first to provide that. I have updated this issue to hopefully explain the state of things better. Sorry for the confusion.

[timostamm]

To elaborate, React Native provides a fetch implementation, but it is incomplete, and does not support streams. Please chime in on facebook/react-native#27741

[timostamm]

We're planning to use it in our app to always get the latest state of IOT devices.

Since there is not a whole lot of activity on the upstream issue: React native can receive streaming text responses via XMLHttpRequest. So if streams are a crucial feature to update your app, and polling is out of the question, it might be worth stepping outside of RPC, and write a bespoke endpoint for this case.

One option is to write a separate endpoint with a simple text-based protocol:

event: somethingchanged
data: somethingid

Every time "something" changed on your backend, you write these two lines of text to the wire, and make sure to flush.

In the client, you read the response body as a text stream, and if you receive the event, you make an RPC call to retrieve the updated information. The data property might contain additional information about what to refresh.

To keep the connection alive, you can write a comment to the wire in an interval, for example:

: ping

The client can ignore this line, but since the data will travel over the wire, it will help to keep the connection alive.

The text protocol we are using here is actually well specified - it's just SSE.

(ping @tantona)

[khalil-omer]

There is a good option for patching fetch in RN here: https://stackoverflow.com/questions/56207968/stream-api-with-fetch-in-a-react-native-app/77089139#77089139.

Any thoughts on documenting this on the Connect side as a way to get this issue addressed despite the upstream issue in RN core?

[smaye81]

Hi @khalil-omer. We are actually aware of that library. Unfortunately, it doesn't help the streaming issue. The details are mentioned in this issue description above. But to summarize:

The library you mention provides text streaming only. The problem is that Connect streaming RPCs are binary. Even with the JSON format, the enveloping is binary. The library's README mentions that if you wish to consume binary data, 'either blob or base64 response types have to be used', but Connect does not implement the gRPC-Web-Text (base64 encoded) format, so this isn't a viable workaround.