And nullable/nonnullable annotation to improve Kotlin API

centrifuge/build.gradle

@@ -26,6 +26,7 @@ dependencies {
     implementation 'com.squareup.okhttp3:okhttp:4.9.3'
     implementation 'com.google.protobuf:protobuf-javalite:3.20.1'
     implementation 'net.sourceforge.streamsupport:streamsupport-cfuture:1.7.0'
+    implementation 'com.google.code.findbugs:jsr305:3.0.2'
 
     testImplementation 'junit:junit:4.13.2'
 }

centrifuge/src/main/java/io/github/centrifugal/centrifuge/Client.java

@@ -17,6 +17,8 @@
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 
+import javax.annotation.Nullable;
+
 import io.github.centrifugal.centrifuge.internal.backoff.Backoff;
 import io.github.centrifugal.centrifuge.internal.protocol.Protocol;
 
@@ -390,7 +392,7 @@ private void handleConnectionOpen() {
                     this.ws.close(NORMAL_CLOSURE_STATUS, "");
                     return;
                 }
-                if (token.equals("")) {
+                if (token == null || token.equals("")) {
                     Client.this.failUnauthorized();
                     return;
                 }
@@ -737,7 +739,7 @@ private void sendRefresh() {
                 );
                 return;
             }
-            if (token.equals("")) {
+            if (token == null || token.equals("")) {
                 this.failUnauthorized();
                 return;
             }
@@ -746,31 +748,22 @@ private void sendRefresh() {
                 if (Client.this.getState() != ClientState.CONNECTED) {
                     return;
                 }
-                if (error != null) {
-                    Client.this.listener.onError(Client.this, new ErrorEvent(new RefreshError(error)));
-                    if (error instanceof ReplyError) {
-                        ReplyError e;
-                        e = (ReplyError) error;
-                        if (e.isTemporary()) {
-                            Client.this.refreshTask = Client.this.scheduler.schedule(
-                                    Client.this::sendRefresh,
-                                    Client.this.backoff.duration(0, 10000, 20000),
-                                    TimeUnit.MILLISECONDS
-                            );
-                        } else {
-                            Client.this.processDisconnect(e.getCode(), e.getMessage(), false);
-                        }
-                        return;
-                    } else {
+                Client.this.listener.onError(Client.this, new ErrorEvent(new RefreshError(error)));
+                if (error instanceof ReplyError) {
+                    ReplyError e;
+                    e = (ReplyError) error;
+                    if (e.isTemporary()) {
                         Client.this.refreshTask = Client.this.scheduler.schedule(
                                 Client.this::sendRefresh,
                                 Client.this.backoff.duration(0, 10000, 20000),
                                 TimeUnit.MILLISECONDS
                         );
+                    } else {
+                        Client.this.processDisconnect(e.getCode(), e.getMessage(), false);
                     }
                     return;
                 }
-                if (result.getExpires()) {
+                if (result != null && result.getExpires()) {
                     int ttl = result.getTtl();
                     Client.this.refreshTask = Client.this.scheduler.schedule(Client.this::sendRefresh, ttl, TimeUnit.SECONDS);
                 }
@@ -1045,11 +1038,11 @@ private void cleanCommandFuture(Protocol.Command cmd) {
      * @param data:   a custom payload for RPC call.
      * @param cb:     will be called as soon as rpc response received or error happened.
      */
-    public void rpc(String method, byte[] data, ResultCallback<RPCResult> cb) {
+    public void rpc(@Nullable String method, byte[] data, ResultCallback<RPCResult> cb) {
         this.executor.submit(() -> Client.this.rpcSynchronized(method, data, cb));
     }
 
-    private void rpcSynchronized(String method, byte[] data, ResultCallback<RPCResult> cb) {
+    private void rpcSynchronized(@Nullable String method, byte[] data, ResultCallback<RPCResult> cb) {
         Protocol.RPCRequest.Builder builder = Protocol.RPCRequest.newBuilder()
                 .setData(com.google.protobuf.ByteString.copyFrom(data));
 

centrifuge/src/main/java/io/github/centrifugal/centrifuge/CompletionCallback.java

@@ -1,8 +1,10 @@
 package io.github.centrifugal.centrifuge;
 
+import javax.annotation.Nullable;
+
 public interface CompletionCallback {
     /**
      * Called when operation done. Caller must check possible error (it's null in case of success).
      */
-    void onDone(Throwable e);
+    void onDone(@Nullable Throwable e);
 }

centrifuge/src/main/java/io/github/centrifugal/centrifuge/RefreshError.java

@@ -1,13 +1,15 @@
 package io.github.centrifugal.centrifuge;
 
+import javax.annotation.Nullable;
+
 public class RefreshError extends Throwable {
     private final Throwable error;
 
-    RefreshError(Throwable error) {
+    RefreshError(@Nullable Throwable error) {
         this.error = error;
     }
 
-    public Throwable getError() {
+    public @Nullable Throwable getError() {
         return error;
     }
 }

centrifuge/src/main/java/io/github/centrifugal/centrifuge/ResultCallback.java

@@ -1,5 +1,7 @@
 package io.github.centrifugal.centrifuge;
 
+import javax.annotation.Nullable;
+
 public interface ResultCallback<T> {
-    void onDone(Throwable e, T result);
+    void onDone(@Nullable Throwable e, @Nullable T result);
 }

centrifuge/src/main/java/io/github/centrifugal/centrifuge/ServerSubscribedEvent.java

@@ -1,7 +1,9 @@
 package io.github.centrifugal.centrifuge;
 
+import javax.annotation.Nullable;
+
 public class ServerSubscribedEvent {
-    ServerSubscribedEvent(String channel, Boolean wasRecovering, Boolean recovered, Boolean positioned, Boolean recoverable, StreamPosition streamPosition, byte[] data) {
+    ServerSubscribedEvent(String channel, Boolean wasRecovering, Boolean recovered, Boolean positioned, Boolean recoverable, @Nullable StreamPosition streamPosition, @Nullable byte[] data) {
         this.channel = channel;
         this.wasRecovering = wasRecovering;
         this.recovered = recovered;

centrifuge/src/main/java/io/github/centrifugal/centrifuge/SubscribedEvent.java

@@ -1,7 +1,9 @@
 package io.github.centrifugal.centrifuge;
 
+import javax.annotation.Nullable;
+
 public class SubscribedEvent {
-    SubscribedEvent(Boolean wasRecovering, Boolean recovered, Boolean positioned, Boolean recoverable, StreamPosition streamPosition, byte[] data) {
+    SubscribedEvent(Boolean wasRecovering, Boolean recovered, Boolean positioned, Boolean recoverable, @Nullable StreamPosition streamPosition, @Nullable byte[] data) {
         this.wasRecovering = wasRecovering;
         this.recovered = recovered;
         this.data = data;

centrifuge/src/main/java/io/github/centrifugal/centrifuge/Subscription.java

@@ -101,7 +101,7 @@ void sendRefresh() {
                 );
                 return;
             }
-            if (token.equals("")) {
+            if (token == null || token.equals("")) {
                 this.failUnauthorized(true);
                 return;
             }
@@ -110,8 +110,9 @@ void sendRefresh() {
                 if (Subscription.this.getState() != SubscriptionState.SUBSCRIBED) {
                     return;
                 }
-                if (error != null) {
-                    Subscription.this.listener.onError(Subscription.this, new SubscriptionErrorEvent(new SubscriptionRefreshError(error)));
+                Throwable errorOrNull = error != null ? error : (result == null ? new NullPointerException() : null);
+                if (errorOrNull != null) {
+                    Subscription.this.listener.onError(Subscription.this, new SubscriptionErrorEvent(new SubscriptionRefreshError(errorOrNull)));
                     if (error instanceof ReplyError) {
                         ReplyError e;
                         e = (ReplyError) error;
@@ -273,7 +274,7 @@ void sendSubscribe() {
                     Subscription.this.scheduleResubscribe();
                     return;
                 }
-                if (token.equals("")) {
+                if (token == null || token.equals("")) {
                     Subscription.this.failUnauthorized(false);
                     return;
                 }

centrifuge/src/main/java/io/github/centrifugal/centrifuge/TokenCallback.java

@@ -1,6 +1,8 @@
 package io.github.centrifugal.centrifuge;
 
+import javax.annotation.Nullable;
+
 public interface TokenCallback {
     /* Call this with exception or token. If called with null error and empty token then client considers access unauthorized */
-    void Done(Throwable e, String token);
+    void Done(@Nullable Throwable e, @Nullable String token);
 }

centrifuge/src/main/java/io/github/centrifugal/centrifuge/package-info.java

@@ -0,0 +1,8 @@
+/**
+ * File: package-info.java
+ * Make all method parameters @NonNull by default
+ */
+@ParametersAreNonnullByDefault
+package io.github.centrifugal.centrifuge;
+
+import javax.annotation.ParametersAreNonnullByDefault;