Update AndroidServerEventsClient to implement interruptible read using OkHttp

Author: mythz

src/AndroidClient/android/src/main/java/net/servicestack/android/AndroidServerEventsClient.java

@@ -5,17 +5,35 @@
 import net.servicestack.client.AsyncResultVoid;
 import net.servicestack.client.AsyncSuccess;
 import net.servicestack.client.AsyncSuccessVoid;
+import net.servicestack.client.Log;
+import net.servicestack.client.sse.EventStream;
 import net.servicestack.client.sse.GetEventSubscribers;
 import net.servicestack.client.sse.ServerEventUser;
 import net.servicestack.client.sse.ServerEventsClient;
 import net.servicestack.client.sse.UpdateEventSubscriber;
 import net.servicestack.client.sse.UpdateEventSubscriberResponse;
 import net.servicestack.func.Func;
 
+import java.io.BufferedInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InterruptedIOException;
+import java.lang.reflect.Field;
+import java.net.HttpURLConnection;
+import java.net.SocketException;
+import java.net.URL;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 
+import okhttp3.Call;
+import okhttp3.Callback;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.Response;
+import okhttp3.ResponseBody;
+import okio.BufferedSource;
+
 /**
  * Created by mythz on 2/15/2017.
  */
@@ -38,6 +56,73 @@ public AndroidServiceClient getAndroidClient(){
         return (AndroidServiceClient) this.serviceClient;
     }
 
+    @Override
+    protected EventStream createEventStream() {
+        return new AndroidEventStream(this);
+    }
+
+    public static final OkHttpClient client = new OkHttpClient();
+
+    class AndroidEventStream extends EventStream
+    {
+        public AndroidEventStream(ServerEventsClient client) {
+            super(client);
+        }
+
+        Call call;
+
+        @Override
+        protected InputStream getInputStream(URL streamUri) throws IOException {
+
+            Request request = new Request.Builder()
+                .url(streamUri)
+                .build();
+
+            call = client.newCall(request);
+            Response response = call.execute();
+
+            return new BufferedInputStream(response.body().byteStream());
+        }
+
+        @Override
+        public void close() {
+            if (call == null) return;
+            call.cancel();
+        }
+
+        @Override
+        protected int readFromStream(InputStream inputStream, byte[] buffer) throws IOException, InterruptedException {
+            int len;
+
+            //Simulates a bgThread.interrupt() so EventStream.run() exits gracefully
+
+            try {
+                 len = inputStream.read(buffer);
+            } catch (SocketException e){
+                if (call.isCanceled()){
+                    throw new InterruptedException();
+                }
+                throw e;
+            }
+            if (call.isCanceled()){
+                throw new InterruptedException();
+            }
+            return len;
+        }
+    }
+
+    @Override
+    protected synchronized void interruptBackgroundThread() {
+        if (bgThread != null){
+            bgEventStream.close();
+            try {
+                bgThread.join(1000);
+            } catch (InterruptedException ignore) {}
+            bgThread = null;
+            bgEventStream = null;
+        }
+    }
+
     public void getChannelSubscribersAsync(final AsyncResult<List<ServerEventUser>> asyncResult){
         this.getAndroidClient().getAsync(
             new GetEventSubscribers().setChannels(Func.toList(this.getChannels())),