load stream throughput via metrics + sock.js instead of polling for t…

…hem every second per stream

issue Graylog2/graylog2-server#1071

app/views/streams/stream.scala.html

@@ -55,7 +55,7 @@ <h2>
         </div>
 
         <div class="stream-metadata">
-            <span class="value node-throughput" data-node-type="stream" data-stream-id="@stream.getId">?</span> messages/second,
+            <span class="react-stream-throughput" data-stream-id="@stream.getId"></span>,
 
             @if(stream.getStreamRules.isEmpty) {
                 no configured rules.

javascript/src/components/streams/StreamThroughput.jsx

@@ -0,0 +1,50 @@
+'use strict';
+
+var React = require('react');
+//noinspection JSUnusedGlobalSymbols
+var MetricsStore = require('../../stores/metrics/MetricsStore');
+
+var metricsStore = MetricsStore.instance;
+
+var StreamThroughput = React.createClass({
+    getInitialState() {
+        return ({
+            throughput: 0,
+            hasError: false
+        });
+    },
+    componentWillMount() {
+        var metricName = "org.graylog2.plugin.streams.Stream." + this.props.streamId + ".incomingMessages.1-sec-rate";
+        metricsStore.listen({
+            nodeId: null, // across all nodes
+            metricNames: [metricName],
+            callback: (update, hasError) => {
+                // update is [{nodeId, values: [{name, value: {metric}}]} ...]
+                // metric can be various different things, depending on metric {type: "GAUGE"|"COUNTER"|"METER"|"TIMER"}
+
+                var throughput = 0;
+                // not using filter.map.reduce because that's even worse to read than this code...
+                update.forEach((perNode) => {
+                    perNode.values.forEach((namedMetric) => {
+                        if (namedMetric.name === metricName) {
+                            throughput += namedMetric.metric.value;
+                        }
+                    });
+                });
+                this.setState({throughput: throughput, hasError: hasError});
+            }
+        });
+    },
+    render() {
+        if (this.state.hasError) {
+            return (<span>Throughput unavailable</span>);
+        }
+
+        return (
+            <span>{this.state.throughput} messages/second</span>
+        );
+
+    }
+});
+
+module.exports = StreamThroughput;

javascript/src/components/streams/mount.jsx

@@ -0,0 +1,13 @@
+'use strict';
+
+var React = require('react/addons');
+var StreamThroughput = require('./StreamThroughput');
+
+var streamThroughput = document.getElementsByClassName('react-stream-throughput');
+if (streamThroughput) {
+    for (var i = 0; i < streamThroughput.length; i++) {
+        var elem = streamThroughput[i];
+        var id = elem.getAttribute('data-stream-id');
+        React.render(<StreamThroughput streamId={id}/>, elem);
+    }
+}

javascript/src/mount.jsx

@@ -17,6 +17,7 @@ $(document).ready(() => {
     require('./components/grok-patterns/mount');
     require('./components/widgets/mount');
     require('./components/throughput/mount');
+    require('./components/streams/mount');
     if (userPreferences.enableSmartSearch) {
         require('./components/search/mount');
     }