migrate dart to null-safety

Author: Marinko Babic

CHANGELOG.md

@@ -1,3 +1,5 @@
+### 1.0.0 - 10/25/2021
+ * NullSafety support
 ### 0.1.2 - 09/21/202
 
  * Add buffer for chunks of one RSocket Frame

justfile

@@ -1,5 +1,5 @@
 format:
- dartfmt -w lib test
+ dart format lib test
 
 server:
  rsocket-cli -i "pong" --server --debug tcp://localhost:42252

lib/core/rsocket_error.dart

@@ -1,5 +1,5 @@
 class RSocketException implements Exception {
- final int code;
+ final int? code;
  final String message;
 
  RSocketException(this.code, this.message);

lib/core/rsocket_requester.dart

@@ -15,7 +15,7 @@ Future<void> voidFuture() async {}
 const MAX_REQUEST_N_SIZE = 0x7FFFFFFF;
 
 abstract class Subscriber {
- void onNext(Payload value);
+ void onNext(Payload? value);
 
  void onError(dynamic error);
 
@@ -24,12 +24,12 @@ abstract class Subscriber {
 
 class CompleterSubscriber implements Subscriber {
  Completer completer;
- Payload payload;
+ Payload? payload;
 
  CompleterSubscriber(this.completer);
 
  @override
- void onNext(Payload payload) {
+ void onNext(Payload? payload) {
  this.payload = payload;
  }
 
@@ -47,11 +47,11 @@ class CompleterSubscriber implements Subscriber {
 class StreamSubscriber implements Subscriber {
  final StreamController controller;
 
- StreamSubscriber({FutureOr<void> onCancel() = null})
+ StreamSubscriber({FutureOr<void> onCancel()? = null})
  : controller = StreamController(onCancel: onCancel);
 
  @override
- void onNext(Payload value) {
+ void onNext(Payload? value) {
  controller.add(value);
  }
 
@@ -65,26 +65,26 @@ class StreamSubscriber implements Subscriber {
  controller.close().then((value) => {});
  }
 
- Stream<Payload> payloadStream() {
- return controller.stream.map((item) => item as Payload);
+ Stream<Payload?> payloadStream() {
+ return controller.stream.map((item) => item as Payload?);
  }
 }
 
 class RSocketRequester extends RSocket {
  bool closed = false;
  double _availability = 1.0;
- Timer keepAliveTimer;
- StreamIdSupplier streamIdSupplier;
- ConnectionSetupPayload connectionSetupPayload;
- DuplexConnection connection;
+ Timer? keepAliveTimer;
+ late StreamIdSupplier streamIdSupplier;
+ ConnectionSetupPayload? connectionSetupPayload;
+ late DuplexConnection connection;
 
  //buffer for data chunk
- List<int> chunkBuffer;
+ List<int>? chunkBuffer;
 
  Map<int, Subscriber> senders = {};
- RSocket responder;
+ RSocket? responder;
  String mode = 'requester';
- ErrorConsumer errorConsumer;
+ ErrorConsumer? errorConsumer;
 
  RSocketRequester(String mode, ConnectionSetupPayload connectionSetupPayload,
  DuplexConnection connection) {
@@ -109,23 +109,23 @@ class RSocketRequester extends RSocket {
  //RSocket requestResponse
  requestResponse = (payload) {
  var completer = Completer<Payload>();
- var streamId = streamIdSupplier.nextStreamId(senders);
+ var streamId = streamIdSupplier.nextStreamId(senders)!;
  connection
- .write(FrameCodec.encodeRequestResponseFrame(streamId, payload));
+ .write(FrameCodec.encodeRequestResponseFrame(streamId, payload!));
  senders[streamId] = CompleterSubscriber(completer);
  return completer.future;
  };
  //RSocket fireAndForget
  fireAndForget = (payload) {
- var streamId = streamIdSupplier.nextStreamId(senders);
- connection.write(FrameCodec.encodeFireAndForgetFrame(streamId, payload));
+ var streamId = streamIdSupplier.nextStreamId(senders)!;
+ connection.write(FrameCodec.encodeFireAndForgetFrame(streamId, payload!));
  return Future.value(() {});
  };
  //RSocket requestStream
  requestStream = (payload) {
- var streamId = streamIdSupplier.nextStreamId(senders);
+ var streamId = streamIdSupplier.nextStreamId(senders)!;
  connection.write(FrameCodec.encodeRequestStreamFrame(
- streamId, MAX_REQUEST_N_SIZE, payload));
+ streamId, MAX_REQUEST_N_SIZE, payload!));
  var streamSubscriber = StreamSubscriber(onCancel: () {
  connection.write(FrameCodec.encodeCancelFrame(streamId));
  senders.remove(streamId);
@@ -135,7 +135,7 @@ class RSocketRequester extends RSocket {
  };
  //RSocket metadataPush
  metadataPush = (payload) {
- connection.write(FrameCodec.encodeMetadataFrame(0, payload));
+ connection.write(FrameCodec.encodeMetadataFrame(0, payload!));
  return Future.value(() {});
  };
  //Rsocket Channel
@@ -153,7 +153,7 @@ class RSocketRequester extends RSocket {
  connection.write(setupPayloadFrame());
  if (mode == 'requester') {
  keepAliveTimer = Timer.periodic(
- Duration(seconds: connectionSetupPayload.keepAliveInterval),
+ Duration(seconds: connectionSetupPayload!.keepAliveInterval),
  (Timer t) {
  if (!closed) {
  connection.write(FrameCodec.encodeKeepAlive(false, 0));
@@ -181,12 +181,12 @@ class RSocketRequester extends RSocket {
 
  void receiveChunk(Uint8List chunk) {
  if (this.chunkBuffer != null) {
- this.chunkBuffer = this.chunkBuffer + chunk;
- var chunkDataLength = this.chunkBuffer.length - 3;
- var bytes = this.chunkBuffer.sublist(0, 3);
- var rsocketFrameLength = bytesToNumber(bytes);
+ this.chunkBuffer = this.chunkBuffer! + chunk;
+ var chunkDataLength = this.chunkBuffer!.length - 3;
+ var bytes = this.chunkBuffer!.sublist(0, 3);
+ var rsocketFrameLength = bytesToNumber(bytes)!;
  if (rsocketFrameLength <= chunkDataLength) {
- for (var frame in parseFrames(this.chunkBuffer)) {
+ for (var frame in parseFrames(this.chunkBuffer!)) {
  receiveFrame(frame);
  }
  this.chunkBuffer = null;
@@ -196,7 +196,7 @@ class RSocketRequester extends RSocket {
  if (chunk.length > 3) {
  var chunkDataLength = chunk.length - 3;
  var bytes = chunk.sublist(0, 3);
- var rsocketFrameLength = bytesToNumber(bytes);
+ var rsocketFrameLength = bytesToNumber(bytes)!;
  if (rsocketFrameLength > chunkDataLength) {
  this.chunkBuffer = chunk;
  return;
@@ -219,12 +219,12 @@ class RSocketRequester extends RSocket {
  if (payloadFrame.completed) {
  senders.remove(streamId);
  if (payload?.data != null) {
- subscriber.onNext(payload);
+ subscriber!.onNext(payload);
  }
- subscriber.onComplete();
+ subscriber!.onComplete();
  } else {
  if (payload?.data != null) {
- subscriber.onNext(payload);
+ subscriber!.onNext(payload);
  }
  }
  }
@@ -241,10 +241,10 @@ class RSocketRequester extends RSocket {
  var streamId = header.streamId;
  var error = RSocketException(errorFrame.code, errorFrame.message);
  if (streamId == 0 && errorConsumer != null) {
- errorConsumer(error);
+ errorConsumer!(error);
  } else {
  if (senders.containsKey(streamId)) {
- var subscriber = senders[streamId];
+ var subscriber = senders[streamId]!;
  senders.remove(streamId);
  subscriber.onError(error);
  }
@@ -261,37 +261,37 @@ class RSocketRequester extends RSocket {
  case frame_types.REQUEST_RESPONSE:
  var requestResponseFrame = frame as RequestResponseFrame;
  if (responder != null && requestResponseFrame.payload != null) {
- responder
- .requestResponse(requestResponseFrame.payload)
+ responder!.requestResponse!(requestResponseFrame.payload)
  .then((payload) {
  connection.write(
  FrameCodec.encodePayloadFrame(header.streamId, true, payload));
  }).catchError((error) {
  var rsocketError = convertToRSocketException(error);
  connection.write(FrameCodec.encodeErrorFrame(
- header.streamId, rsocketError.code, rsocketError.message));
+ header.streamId, rsocketError.code!, rsocketError.message));
  });
  }
  break;
  case frame_types.REQUEST_FNF:
  var fireAndForgetFrame = frame as RequestFNFFrame;
  if (responder != null && fireAndForgetFrame.payload != null) {
- responder
- .fireAndForget(fireAndForgetFrame.payload)
+ responder!.fireAndForget!(fireAndForgetFrame.payload)
  .then((value) => {});
  }
  break;
  case frame_types.METADATA_PUSH:
  var metadataPushFrame = frame as MetadataPushFrame;
  if (responder != null && metadataPushFrame.payload != null) {
- responder.metadataPush(metadataPushFrame.payload).then((value) => {});
+ responder!.metadataPush!(metadataPushFrame.payload)
+ .then((value) => {});
  }
  break;
  case frame_types.REQUEST_STREAM:
  var requestStreamFrame = frame as RequestStreamFrame;
  var requesterStreamId = header.streamId;
  if (responder != null && requestStreamFrame.payload != null) {
- responder.requestStream(requestStreamFrame.payload).listen((payload) {
+ responder!.requestStream!(requestStreamFrame.payload).listen(
+ (payload) {
  connection.write(FrameCodec.encodePayloadFrame(
  requesterStreamId, false, payload));
  }, onDone: () {
@@ -301,7 +301,7 @@ class RSocketRequester extends RSocket {
  if (error is RSocketException) {
  var e = error;
  connection.write(FrameCodec.encodeErrorFrame(
- requesterStreamId, e.code, e.message));
+ requesterStreamId, e.code!, e.message));
  } else {
  connection.write(FrameCodec.encodeErrorFrame(requesterStreamId,
  RSocketErrorCode.APPLICATION_ERROR, error.toString()));
@@ -315,10 +315,10 @@ class RSocketRequester extends RSocket {
 
  Uint8List setupPayloadFrame() {
  return FrameCodec.encodeSetupFrame(
- connectionSetupPayload.keepAliveInterval,
- connectionSetupPayload.keepAliveMaxLifetime,
- connectionSetupPayload.metadataMimeType,
- connectionSetupPayload.dataMimeType,
+ connectionSetupPayload!.keepAliveInterval,
+ connectionSetupPayload!.keepAliveMaxLifetime,
+ connectionSetupPayload!.metadataMimeType,
+ connectionSetupPayload!.dataMimeType,
  connectionSetupPayload);
  }
 }

lib/core/rsocket_responder.dart

@@ -8,11 +8,11 @@ import '../payload.dart';
 import '../rsocket.dart';
 
 class BaseResponder {
- SocketAcceptor socketAcceptor;
- Uri uri;
+ late SocketAcceptor socketAcceptor;
+ late Uri uri;
 
  Future<void> receiveConnection(DuplexConnection duplexConn) async {
- RSocketRequester rsocketRequester;
+ RSocketRequester? rsocketRequester;
  duplexConn.receiveHandler = (chunk) {
  for (var frame in parseFrames(chunk)) {
  var header = frame.header;
@@ -45,7 +45,7 @@ class BaseResponder {
 }
 
 class TcpRSocketResponder extends BaseResponder implements Closeable {
- ServerSocket serverSocket;
+ late ServerSocket serverSocket;
 
  TcpRSocketResponder(
  Uri uri, ServerSocket serverSocket, SocketAcceptor socketAcceptor) {
@@ -67,7 +67,7 @@ class TcpRSocketResponder extends BaseResponder implements Closeable {
 }
 
 class WebSocketRSocketResponder extends BaseResponder implements Closeable {
- HttpServer httpServer;
+ late HttpServer httpServer;
 
  WebSocketRSocketResponder(
  Uri uri, HttpServer httpServer, SocketAcceptor socketAcceptor) {

lib/core/stream_id_supplier.dart

@@ -1,7 +1,7 @@
 class StreamIdSupplier {
  static int MASK = 0x7FFFFFFF;
- int streamId;
- int initialValue;
+ int streamId = 0;
+ late int initialValue;
 
  StreamIdSupplier();
 
@@ -18,7 +18,7 @@ class StreamIdSupplier {
  return StreamIdSupplier.streamId(0);
  }
 
- int nextStreamId(Map<int, dynamic> streamIds) {
+ int? nextStreamId(Map<int, dynamic> streamIds) {
  var nextStreamId;
  do {
  streamId += 2;

lib/duplex_connection.dart

@@ -1,14 +1,13 @@
 import 'dart:io';
 import 'dart:typed_data';
 
-import 'rsocket.dart';
-
 import 'io/bytes.dart';
+import 'rsocket.dart';
 
 abstract class DuplexConnection implements Closeable, Availability {
  double _availability = 1.0;
- TcpChunkHandler receiveHandler;
- CloseHandler closeHandler;
+ TcpChunkHandler? receiveHandler;
+ CloseHandler? closeHandler;
 
  void init();
 
@@ -32,7 +31,7 @@ class TcpDuplexConnection extends DuplexConnection {
  @override
  void init() {
  socket.listen((data) {
- receiveHandler(data);
+ receiveHandler!(data);
  }, onDone: () {
  close();
  }, onError: (e) {
@@ -45,12 +44,8 @@ class TcpDuplexConnection extends DuplexConnection {
  if (!closed) {
  closed = true;
  _availability = 0.0;
- if (socket != null) {
- socket.close();
- }
- if (closeHandler != null) {
- closeHandler();
- }
+ socket.close();
+ closeHandler?.call();
  }
  }
 
@@ -71,7 +66,7 @@ class WebSocketDuplexConnection extends DuplexConnection {
  webSocket.listen((message) {
  var data = message as List<int>;
  var frameLenBytes = i24ToBytes(data.length);
- receiveHandler(Uint8List.fromList(frameLenBytes + data));
+ receiveHandler!(Uint8List.fromList(frameLenBytes + data));
  }, onDone: () {
  close();
  }, onError: (e) {
@@ -84,12 +79,8 @@ class WebSocketDuplexConnection extends DuplexConnection {
  if (!closed) {
  closed = true;
  _availability = 0.0;
- if (webSocket != null) {
- webSocket.close();
- }
- if (closeHandler != null) {
- closeHandler();
- }
+ webSocket.close();
+ closeHandler?.call();
  }
  }