provides extra @NonNullApi annotation for all packages

Signed-off-by: Oleh Dokuka <[email protected]>

Author: OlegDokuka

rsocket-core/src/main/java/io/rsocket/ConnectionSetupPayload.java

@@ -19,7 +19,7 @@
 import io.netty.buffer.ByteBuf;
 import io.netty.util.AbstractReferenceCounted;
 import io.rsocket.core.DefaultConnectionSetupPayload;
-import javax.annotation.Nullable;
+import reactor.util.annotation.Nullable;
 
 /**
  * Exposes information from the {@code SETUP} frame to a server, as well as to client responders.

rsocket-core/src/main/java/io/rsocket/core/RSocketConnector.java

@@ -41,10 +41,10 @@
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
-import javax.annotation.Nullable;
 import reactor.core.Disposable;
 import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Schedulers;
+import reactor.util.annotation.Nullable;
 import reactor.util.retry.Retry;
 
 /**

rsocket-core/src/main/java/io/rsocket/core/RSocketRequester.java

@@ -55,7 +55,6 @@
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
-import javax.annotation.Nullable;
 import org.reactivestreams.Processor;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
@@ -68,6 +67,7 @@
 import reactor.core.publisher.SignalType;
 import reactor.core.publisher.UnicastProcessor;
 import reactor.core.scheduler.Scheduler;
+import reactor.util.annotation.Nullable;
 import reactor.util.concurrent.Queues;
 
 /**

rsocket-core/src/main/java/io/rsocket/core/RSocketResponder.java

@@ -39,14 +39,14 @@
 import java.util.function.Consumer;
 import java.util.function.LongConsumer;
 import java.util.function.Supplier;
-import javax.annotation.Nullable;
 import org.reactivestreams.Processor;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 import reactor.core.Disposable;
 import reactor.core.Exceptions;
 import reactor.core.publisher.*;
+import reactor.util.annotation.Nullable;
 import reactor.util.concurrent.Queues;
 
 /** Responder side of RSocket. Receives {@link ByteBuf}s from a peer's {@link RSocketRequester} */

rsocket-core/src/main/java/io/rsocket/core/StreamIdSupplier.java

@@ -16,7 +16,9 @@
 package io.rsocket.core;
 
 import io.netty.util.collection.IntObjectMap;
+import javax.annotation.concurrent.NotThreadSafe;
 
+@NotThreadSafe
 final class StreamIdSupplier {
   private static final int MASK = 0x7FFFFFFF;
 

rsocket-core/src/main/java/io/rsocket/exceptions/ApplicationErrorException.java

@@ -17,7 +17,7 @@
 package io.rsocket.exceptions;
 
 import io.rsocket.frame.ErrorFrameCodec;
-import javax.annotation.Nullable;
+import reactor.util.annotation.Nullable;
 
 /**
  * Application layer logic generating a Reactive Streams {@code onError} event.

rsocket-core/src/main/java/io/rsocket/exceptions/CanceledException.java

@@ -17,7 +17,7 @@
 package io.rsocket.exceptions;
 
 import io.rsocket.frame.ErrorFrameCodec;
-import javax.annotation.Nullable;
+import reactor.util.annotation.Nullable;
 
 /**
  * The Responder canceled the request but may have started processing it (similar to REJECTED but

rsocket-core/src/main/java/io/rsocket/exceptions/ConnectionCloseException.java

@@ -17,7 +17,7 @@
 package io.rsocket.exceptions;
 
 import io.rsocket.frame.ErrorFrameCodec;
-import javax.annotation.Nullable;
+import reactor.util.annotation.Nullable;
 
 /**
  * The connection is being terminated. Sender or Receiver of this frame MUST wait for outstanding

rsocket-core/src/main/java/io/rsocket/exceptions/ConnectionErrorException.java

@@ -17,7 +17,7 @@
 package io.rsocket.exceptions;
 
 import io.rsocket.frame.ErrorFrameCodec;
-import javax.annotation.Nullable;
+import reactor.util.annotation.Nullable;
 
 /**
  * The connection is being terminated. Sender or Receiver of this frame MAY close the connection

rsocket-core/src/main/java/io/rsocket/exceptions/CustomRSocketException.java

@@ -17,7 +17,7 @@
 package io.rsocket.exceptions;
 
 import io.rsocket.frame.ErrorFrameCodec;
-import javax.annotation.Nullable;
+import reactor.util.annotation.Nullable;
 
 public class CustomRSocketException extends RSocketException {
   private static final long serialVersionUID = 7873267740343446585L;

rsocket-core/src/main/java/io/rsocket/exceptions/InvalidException.java

@@ -17,7 +17,7 @@
 package io.rsocket.exceptions;
 
 import io.rsocket.frame.ErrorFrameCodec;
-import javax.annotation.Nullable;
+import reactor.util.annotation.Nullable;
 
 /**
  * The request is invalid.

rsocket-core/src/main/java/io/rsocket/exceptions/InvalidSetupException.java

@@ -17,7 +17,7 @@
 package io.rsocket.exceptions;
 
 import io.rsocket.frame.ErrorFrameCodec;
-import javax.annotation.Nullable;
+import reactor.util.annotation.Nullable;
 
 /**
  * The Setup frame is invalid for the server (it could be that the client is too recent for the old

rsocket-core/src/main/java/io/rsocket/exceptions/RejectedException.java

@@ -17,7 +17,7 @@
 package io.rsocket.exceptions;
 
 import io.rsocket.frame.ErrorFrameCodec;
-import javax.annotation.Nullable;
+import reactor.util.annotation.Nullable;
 
 /**
  * Despite being a valid request, the Responder decided to reject it. The Responder guarantees that

rsocket-core/src/main/java/io/rsocket/exceptions/RejectedResumeException.java

@@ -17,7 +17,7 @@
 package io.rsocket.exceptions;
 
 import io.rsocket.frame.ErrorFrameCodec;
-import javax.annotation.Nullable;
+import reactor.util.annotation.Nullable;
 
 /**
  * The server rejected the resume, it can specify the reason in the payload.

rsocket-core/src/main/java/io/rsocket/exceptions/RejectedSetupException.java

@@ -17,7 +17,7 @@
 package io.rsocket.exceptions;
 
 import io.rsocket.frame.ErrorFrameCodec;
-import javax.annotation.Nullable;
+import reactor.util.annotation.Nullable;
 
 /**
  * The server rejected the setup, it can specify the reason in the payload.

rsocket-core/src/main/java/io/rsocket/exceptions/SetupException.java

@@ -17,7 +17,7 @@
 package io.rsocket.exceptions;
 
 import io.rsocket.frame.ErrorFrameCodec;
-import javax.annotation.Nullable;
+import reactor.util.annotation.Nullable;
 
 /** The root of the setup exception hierarchy. */
 public abstract class SetupException extends RSocketException {

rsocket-core/src/main/java/io/rsocket/exceptions/UnsupportedSetupException.java

@@ -17,7 +17,7 @@
 package io.rsocket.exceptions;
 
 import io.rsocket.frame.ErrorFrameCodec;
-import javax.annotation.Nullable;
+import reactor.util.annotation.Nullable;
 
 /**
  * Some (or all) of the parameters specified by the client are unsupported by the server.

rsocket-core/src/main/java/io/rsocket/fragmentation/FragmentationDuplexConnection.java

@@ -25,12 +25,12 @@
 import io.rsocket.frame.FrameLengthCodec;
 import io.rsocket.frame.FrameType;
 import java.util.Objects;
-import javax.annotation.Nullable;
 import org.reactivestreams.Publisher;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.util.annotation.Nullable;
 
 /**
  * A {@link DuplexConnection} implementation that fragments and reassembles {@link ByteBuf}s.

rsocket-core/src/main/java/io/rsocket/fragmentation/FrameReassembler.java

@@ -29,6 +29,7 @@
 import org.slf4j.LoggerFactory;
 import reactor.core.Disposable;
 import reactor.core.publisher.SynchronousSink;
+import reactor.util.annotation.Nullable;
 
 /**
  * The implementation of the RSocket reassembly behavior.
@@ -83,6 +84,7 @@ public boolean isDisposed() {
     return get();
   }
 
+  @Nullable
   synchronized ByteBuf getHeader(int streamId) {
     return headers.get(streamId);
   }
@@ -109,14 +111,17 @@ synchronized CompositeByteBuf getData(int streamId) {
     return byteBuf;
   }
 
+  @Nullable
   synchronized ByteBuf removeHeader(int streamId) {
     return headers.remove(streamId);
   }
 
+  @Nullable
   synchronized CompositeByteBuf removeMetadata(int streamId) {
     return metadata.remove(streamId);
   }
 
+  @Nullable
   synchronized CompositeByteBuf removeData(int streamId) {
     return data.remove(streamId);
   }
@@ -236,7 +241,6 @@ void reassembleFrame(ByteBuf frame, SynchronousSink<ByteBuf> sink) {
         case CANCEL:
         case ERROR:
           cancelAssemble(streamId);
-        default:
       }
 
       if (!frameType.isFragmentable()) {
@@ -270,7 +274,7 @@ private ByteBuf assembleFrameWithMetadata(ByteBuf frame, int streamId, ByteBuf h
         metadata = PayloadFrameCodec.metadata(frame).retain();
       }
     } else {
-      metadata = cm != null ? cm : null;
+      metadata = cm;
     }
 
     ByteBuf data = assembleData(frame, streamId);

rsocket-core/src/main/java/io/rsocket/frame/ExtensionFrameCodec.java

@@ -2,7 +2,7 @@
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
-import javax.annotation.Nullable;
+import reactor.util.annotation.Nullable;
 
 public class ExtensionFrameCodec {
   private ExtensionFrameCodec() {}
@@ -49,6 +49,7 @@ public static ByteBuf data(ByteBuf byteBuf) {
     return data;
   }
 
+  @Nullable
   public static ByteBuf metadata(ByteBuf byteBuf) {
     FrameHeaderCodec.ensureFrameType(FrameType.EXT, byteBuf);
 

rsocket-core/src/main/java/io/rsocket/frame/FrameBodyCodec.java

@@ -3,6 +3,7 @@
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.Unpooled;
+import reactor.util.annotation.Nullable;
 
 class FrameBodyCodec {
   public static final int FRAME_LENGTH_MASK = 0xFFFFFF;
@@ -33,9 +34,9 @@ private static int decodeLength(final ByteBuf byteBuf) {
   static ByteBuf encode(
       ByteBufAllocator allocator,
       final ByteBuf header,
-      ByteBuf metadata,
+      @Nullable ByteBuf metadata,
       boolean hasMetadata,
-      ByteBuf data) {
+      @Nullable ByteBuf data) {
 
     final boolean addData;
     if (data != null) {

rsocket-core/src/main/java/io/rsocket/frame/GenericFrameCodec.java

@@ -4,7 +4,7 @@
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.util.IllegalReferenceCountException;
 import io.rsocket.Payload;
-import javax.annotation.Nullable;
+import reactor.util.annotation.Nullable;
 
 class GenericFrameCodec {
 
@@ -75,7 +75,7 @@ static ByteBuf encode(
       boolean next,
       int requestN,
       @Nullable ByteBuf metadata,
-      ByteBuf data) {
+      @Nullable ByteBuf data) {
 
     final boolean hasMetadata = metadata != null;
 
@@ -115,6 +115,7 @@ static ByteBuf data(ByteBuf byteBuf) {
     return data;
   }
 
+  @Nullable
   static ByteBuf metadata(ByteBuf byteBuf) {
     boolean hasMetadata = FrameHeaderCodec.hasMetadata(byteBuf);
     if (!hasMetadata) {
@@ -136,6 +137,7 @@ static ByteBuf dataWithRequestN(ByteBuf byteBuf) {
     return data;
   }
 
+  @Nullable
   static ByteBuf metadataWithRequestN(ByteBuf byteBuf) {
     boolean hasMetadata = FrameHeaderCodec.hasMetadata(byteBuf);
     if (!hasMetadata) {

rsocket-core/src/main/java/io/rsocket/frame/LeaseFrameCodec.java

@@ -2,8 +2,7 @@
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
-import io.netty.buffer.Unpooled;
-import javax.annotation.Nullable;
+import reactor.util.annotation.Nullable;
 
 public class LeaseFrameCodec {
 
@@ -67,6 +66,7 @@ public static int numRequests(final ByteBuf byteBuf) {
     return numRequests;
   }
 
+  @Nullable
   public static ByteBuf metadata(final ByteBuf byteBuf) {
     FrameHeaderCodec.ensureFrameType(FrameType.LEASE, byteBuf);
     if (FrameHeaderCodec.hasMetadata(byteBuf)) {
@@ -77,7 +77,7 @@ public static ByteBuf metadata(final ByteBuf byteBuf) {
       byteBuf.resetReaderIndex();
       return metadata;
     } else {
-      return Unpooled.EMPTY_BUFFER;
+      return null;
     }
   }
 }

rsocket-core/src/main/java/io/rsocket/frame/MetadataPushFrameCodec.java

@@ -8,6 +8,10 @@
 public class MetadataPushFrameCodec {
 
   public static ByteBuf encodeReleasingPayload(ByteBufAllocator allocator, Payload payload) {
+    if (!payload.hasMetadata()) {
+      throw new IllegalStateException(
+          "Metadata  push requires to have metadata present" + " in the given Payload");
+    }
     final ByteBuf metadata = payload.metadata().retain();
     // releasing payload safely since it can be already released wheres we have to release retained
     // data and metadata as well

rsocket-core/src/main/java/io/rsocket/frame/PayloadFrameCodec.java

@@ -3,6 +3,7 @@
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
 import io.rsocket.Payload;
+import reactor.util.annotation.Nullable;
 
 public class PayloadFrameCodec {
 
@@ -37,8 +38,8 @@ public static ByteBuf encode(
       boolean fragmentFollows,
       boolean complete,
       boolean next,
-      ByteBuf metadata,
-      ByteBuf data) {
+      @Nullable ByteBuf metadata,
+      @Nullable ByteBuf data) {
 
     return GenericFrameCodec.encode(
         allocator, FrameType.PAYLOAD, streamId, fragmentFollows, complete, next, 0, metadata, data);
@@ -48,6 +49,7 @@ public static ByteBuf data(ByteBuf byteBuf) {
     return GenericFrameCodec.data(byteBuf);
   }
 
+  @Nullable
   public static ByteBuf metadata(ByteBuf byteBuf) {
     return GenericFrameCodec.metadata(byteBuf);
   }

rsocket-core/src/main/java/io/rsocket/frame/RequestChannelFrameCodec.java

@@ -3,6 +3,7 @@
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
 import io.rsocket.Payload;
+import reactor.util.annotation.Nullable;
 
 public class RequestChannelFrameCodec {
 
@@ -31,7 +32,7 @@ public static ByteBuf encode(
       boolean fragmentFollows,
       boolean complete,
       long initialRequestN,
-      ByteBuf metadata,
+      @Nullable ByteBuf metadata,
       ByteBuf data) {
 
     if (initialRequestN < 1) {
@@ -56,6 +57,7 @@ public static ByteBuf data(ByteBuf byteBuf) {
     return GenericFrameCodec.dataWithRequestN(byteBuf);
   }
 
+  @Nullable
   public static ByteBuf metadata(ByteBuf byteBuf) {
     return GenericFrameCodec.metadataWithRequestN(byteBuf);
   }