ReactiveSocket metrics client/server

hystrix-contrib/hystrix-reactivesocket-event-stream/build.gradle

@@ -9,16 +9,16 @@ targetCompatibility = JavaVersion.VERSION_1_8
 
 dependencies {
     compile project(':hystrix-core')
-
-    compile 'io.reactivex:rxjava-reactive-streams:latest.release'
-
+
     compile 'com.fasterxml.jackson.core:jackson-core:latest.release'
     compile 'com.fasterxml.jackson.core:jackson-databind:latest.release'
     compile 'com.fasterxml.jackson.core:jackson-annotations:latest.release'
     compile 'com.fasterxml.jackson.module:jackson-module-afterburner:latest.release'
     compile 'com.fasterxml.jackson.dataformat:jackson-dataformat-cbor:latest.release'
     compile 'io.reactivesocket:reactivesocket:latest.release'
-
+	compile 'io.reactivesocket:reactivesocket-netty:latest.release'
+    compile 'io.reactivex:rxjava-reactive-streams:latest.release'
+
     testCompile 'junit:junit-dep:4.10'
     testCompile 'org.mockito:mockito-all:1.9.5'
 }

hystrix-contrib/hystrix-reactivesocket-event-stream/src/main/java/com/netflix/hystrix/contrib/reactivesocket/EventStream.java

@@ -0,0 +1,92 @@
+/**
+ * Copyright 2016 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.netflix.hystrix.contrib.reactivesocket;
+
+import com.netflix.hystrix.config.HystrixConfigurationStream;
+import com.netflix.hystrix.contrib.reactivesocket.serialize.SerialHystrixConfiguration;
+import com.netflix.hystrix.contrib.reactivesocket.serialize.SerialHystrixDashboardData;
+import com.netflix.hystrix.contrib.reactivesocket.serialize.SerialHystrixMetric;
+import com.netflix.hystrix.contrib.reactivesocket.serialize.SerialHystrixRequestEvents;
+import com.netflix.hystrix.contrib.reactivesocket.serialize.SerialHystrixUtilization;
+import com.netflix.hystrix.metric.HystrixRequestEventsStream;
+import com.netflix.hystrix.metric.sample.HystrixUtilizationStream;
+import io.reactivesocket.Payload;
+import rx.Observable;
+
+import java.util.Arrays;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
+
+class EventStream implements Supplier<Observable<Payload>> {
+
+    private final static int CONFIGURATION_DATA_INTERVAL_IN_MS = 500;
+    private final static int UTILIZATION_DATA_INTERVAL_IN_MS = 500;
+
+    private final Observable<Payload> source;
+    private final AtomicBoolean isSourceCurrentlySubscribed = new AtomicBoolean(false);
+
+    /* package-private */EventStream(Observable<Payload> source) {
+        this.source = source
+                .doOnSubscribe(() -> isSourceCurrentlySubscribed.set(true))
+                .doOnUnsubscribe(() -> isSourceCurrentlySubscribed.set(false))
+                .share()
+                .onBackpressureDrop();
+    }
+
+    @Override
+    public Observable<Payload> get() {
+        return source;
+    }
+
+    public static EventStream getInstance(EventStreamEnum eventStreamEnum) {
+        final Observable<Payload> source;
+
+        switch (eventStreamEnum) {
+            case CONFIG_STREAM:
+                source = new HystrixConfigurationStream(CONFIGURATION_DATA_INTERVAL_IN_MS)
+                        .observe()
+                        .map(SerialHystrixConfiguration::toBytes)
+                        .map(SerialHystrixMetric::toPayload);
+                break;
+            case REQUEST_EVENT_STREAM:
+                source = HystrixRequestEventsStream.getInstance()
+                        .observe()
+                        .map(SerialHystrixRequestEvents::toBytes)
+                        .map(SerialHystrixMetric::toPayload);
+                break;
+            case UTILIZATION_STREAM:
+                source = new HystrixUtilizationStream(UTILIZATION_DATA_INTERVAL_IN_MS)
+                        .observe()
+                        .map(SerialHystrixUtilization::toBytes)
+                        .map(SerialHystrixMetric::toPayload);
+                break;
+            case GENERAL_DASHBOARD_STREAM:
+                source = HystrixDashboardStream.getInstance()
+                        .observe()
+                        .map(SerialHystrixDashboardData::toBytes)
+                        .map(SerialHystrixMetric::toPayload);
+                break;
+            default:
+                throw new IllegalArgumentException("Unknown EventStreamEnum : " + eventStreamEnum);
+        }
+
+        return new EventStream(source);
+    }
+
+    public boolean isSourceCurrentlySubscribed() {
+        return isSourceCurrentlySubscribed.get();
+    }
+}

hystrix-contrib/hystrix-reactivesocket-event-stream/src/main/java/com/netflix/hystrix/contrib/reactivesocket/EventStreamEnum.java

@@ -1,71 +1,39 @@
+/**
+ * Copyright 2016 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 package com.netflix.hystrix.contrib.reactivesocket;
 
-
-import com.netflix.hystrix.contrib.reactivesocket.metrics.HystrixCollapserMetricsStream;
-import com.netflix.hystrix.contrib.reactivesocket.metrics.HystrixCommandMetricsStream;
-import com.netflix.hystrix.contrib.reactivesocket.metrics.HystrixThreadPoolMetricsStream;
-import com.netflix.hystrix.contrib.reactivesocket.requests.HystrixRequestEventsStream;
-import com.netflix.hystrix.contrib.reactivesocket.sample.HystrixConfigStream;
-import com.netflix.hystrix.contrib.reactivesocket.sample.HystrixUtilizationStream;
-import io.reactivesocket.Payload;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import rx.Observable;
-
 import java.util.Arrays;
-import java.util.function.Supplier;
-
-public enum EventStreamEnum implements Supplier<Observable<Payload>> {
-
-    CONFIG_STREAM(1) {
-        @Override
-        public Observable<Payload> get() {
-            logger.info("streaming config data");
-            return HystrixConfigStream.getInstance().get();
-        }
-    },
-    REQUEST_EVENT_STREAM(2) {
-        @Override
-        public Observable<Payload> get() {
-            logger.info("streaming request events");
-            return HystrixRequestEventsStream.getInstance().get();
-        }
-    },
-    UTILIZATION_EVENT_STREAM(3) {
-        @Override
-        public Observable<Payload> get() {
-            logger.info("streaming utilization events");
-            return HystrixUtilizationStream.getInstance().get();
-        }
-    },
-    METRICS_STREAM(4) {
-        @Override
-        public Observable<Payload> get() {
-            logger.info("streaming metrics");
-            return Observable.merge(
-                    HystrixCommandMetricsStream.getInstance().get(),
-                    HystrixThreadPoolMetricsStream.getInstance().get(),
-                    HystrixCollapserMetricsStream.getInstance().get());
-        }
-    }
 
-    ;
+public enum EventStreamEnum {
 
-    private static final Logger logger = LoggerFactory.getLogger(EventStreamEnum.class);
+    CONFIG_STREAM(1), REQUEST_EVENT_STREAM(2), UTILIZATION_STREAM(3), GENERAL_DASHBOARD_STREAM(4);
 
-    private int typeId;
+    private final int typeId;
 
     EventStreamEnum(int typeId) {
         this.typeId = typeId;
     }
 
     public static EventStreamEnum findByTypeId(int typeId) {
         return Arrays
-            .asList(EventStreamEnum.values())
-            .stream()
-            .filter(t -> t.typeId == typeId)
-            .findAny()
-            .orElseThrow(() -> new IllegalStateException("no type id found for id => " + typeId));
+                .asList(EventStreamEnum.values())
+                .stream()
+                .filter(t -> t.typeId == typeId)
+                .findAny()
+                .orElseThrow(() -> new IllegalStateException("no type id found for id => " + typeId));
     }
 
     public int getTypeId() {

hystrix-contrib/hystrix-reactivesocket-event-stream/src/main/java/com/netflix/hystrix/contrib/reactivesocket/EventStreamRequestHandler.java

@@ -1,7 +1,23 @@
+/**
+ * Copyright 2016 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 package com.netflix.hystrix.contrib.reactivesocket;
 
 import io.reactivesocket.Payload;
 import io.reactivesocket.RequestHandler;
+import org.agrona.BitUtil;
 import org.reactivestreams.Publisher;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -18,35 +34,56 @@ public class EventStreamRequestHandler extends RequestHandler {
 
     @Override
     public Publisher<Payload> handleRequestResponse(Payload payload) {
-        return NO_REQUEST_RESPONSE_HANDLER.apply(payload);
+        Observable<Payload> singleResponse = Observable.defer(() -> {
+            try {
+                int typeId = payload.getData().getInt(0);
+                EventStreamEnum eventStreamEnum = EventStreamEnum.findByTypeId(typeId);
+                EventStream eventStream = EventStream.getInstance(eventStreamEnum);
+                return eventStream.get().take(1);
+            } catch (Throwable t) {
+                logger.error(t.getMessage(), t);
+                return Observable.error(t);
+            }
+        });
+
+        return RxReactiveStreams.toPublisher(singleResponse);
     }
 
     @Override
     public Publisher<Payload> handleRequestStream(Payload payload) {
-        return NO_REQUEST_STREAM_HANDLER.apply(payload);
+        Observable<Payload> multiResponse = Observable.defer(() -> {
+            try {
+                int typeId = payload.getData().getInt(0);
+                int numRequested = payload.getData().getInt(BitUtil.SIZE_OF_INT);
+                EventStreamEnum eventStreamEnum = EventStreamEnum.findByTypeId(typeId);
+                EventStream eventStream = EventStream.getInstance(eventStreamEnum);
+                return eventStream.get().take(numRequested);
+            } catch (Throwable t) {
+                logger.error(t.getMessage(), t);
+                return Observable.error(t);
+            }
+        });
+
+        return RxReactiveStreams.toPublisher(multiResponse);
     }
 
     @Override
     public Publisher<Payload> handleSubscription(Payload payload) {
-        Observable<Payload> defer = Observable
+        Observable<Payload> infiniteResponse = Observable
             .defer(() -> {
                 try {
-                    int typeId = payload
-                        .getData()
-                        .getInt(0);
-
+                    int typeId = payload.getData().getInt(0);
                     EventStreamEnum eventStreamEnum = EventStreamEnum.findByTypeId(typeId);
-                    return eventStreamEnum
-                        .get();
+                    EventStream eventStream = EventStream.getInstance(eventStreamEnum);
+                    return eventStream.get();
                 } catch (Throwable t) {
                     logger.error(t.getMessage(), t);
                     return Observable.error(t);
                 }
             })
             .onBackpressureDrop();
 
-        return RxReactiveStreams
-            .toPublisher(defer);
+        return RxReactiveStreams.toPublisher(infiniteResponse);
     }
 
     @Override