From b7900724be1fc440691ff28484abda1d2fd6d5d8 Mon Sep 17 00:00:00 2001
From: Radovan Zvoncek <radovan@thelastpickle.com>
Date: Thu, 5 Jul 2018 15:28:59 +0300
Subject: [PATCH] Reaper #369 Frontend for showing node streams

---
 src/ui/app/jsx/node-status.jsx |   4 ++
 src/ui/app/jsx/stream.jsx      |  66 +++++++++++++++++++++
 src/ui/app/jsx/streams.jsx     | 104 +++++++++++++++++++++++++++++++++
 3 files changed, 174 insertions(+)
 create mode 100644 src/ui/app/jsx/stream.jsx
 create mode 100644 src/ui/app/jsx/streams.jsx

diff --git a/src/ui/app/jsx/node-status.jsx b/src/ui/app/jsx/node-status.jsx
index 7023d1212..1fa38a5fc 100644
--- a/src/ui/app/jsx/node-status.jsx
+++ b/src/ui/app/jsx/node-status.jsx
@@ -1,6 +1,7 @@
 import React from "react";
 import Snapshot from "jsx/snapshot";
 import TpStats from "jsx/tpstats";
+import Streams from "jsx/streams";
 import DroppedMessages from "jsx/dropped-messages";
 import ClientRequestLatency from "jsx/client-request-latency";
 import {DeleteStatusMessageMixin, humanFileSize, getUrlPrefix, toast} from "jsx/mixin";
@@ -249,6 +250,9 @@ const NodeStatus = React.createClass({
                         </div>
                       </div>
                       </Tab>
+                      <Tab eventKey={5} title="Streams">
+                          <Streams endpoint={this.props.endpointStatus.endpoint} clusterName={this.props.clusterName}/>
+                      </Tab>
                     </Tabs>               
                     </div>
                     </div>
diff --git a/src/ui/app/jsx/stream.jsx b/src/ui/app/jsx/stream.jsx
new file mode 100644
index 000000000..249af2a16
--- /dev/null
+++ b/src/ui/app/jsx/stream.jsx
@@ -0,0 +1,66 @@
+import React from "react";
+import ProgressBar from 'react-bootstrap/lib/ProgressBar';
+import Table from 'react-bootstrap/lib/Table';
+import {DeleteStatusMessageMixin, humanFileSize, getUrlPrefix, toast} from "jsx/mixin";
+
+const Stream = React.createClass({
+    propTypes: {
+        planId: React.PropTypes.string.isRequired,
+        direction: React.PropTypes.string.isRequired,
+        stream: React.PropTypes.object.isRequired
+    },
+
+    getInitialState() {
+        return {communicating: false, collapsed: true};
+    },
+
+
+    render: function() {
+
+        const stream = this.props.stream;
+        const isActive = stream.completed ? false : true;
+        const style = stream.success ? (stream.completed ? "success" : "info") : "danger";
+        const state = stream.success ? (stream.completed ? "Done" : "Streaming") : "Error";
+
+        if (this.props.direction == "incoming") {
+            var progress = stream.sizeReceived / stream.sizeToReceive * 100;
+            p = humanFileSize(stream.sizeReceived, 1024) + " / " + humanFileSize(stream.sizeToReceive, 1024);
+            var label = state + " [ " + p + " ]"
+            var tables = Object.values(stream.progressReceived).map(tableProgress => tableProgress.table);
+            var directionText = "From: ";
+        };
+
+        if (this.props.direction == "outgoing") {
+            var progress = stream.sizeSent / stream.sizeToSend * 100;
+            var p = humanFileSize(stream.sizeSent, 1024) + " / " + humanFileSize(stream.sizeToSend, 1024);
+            var label = state + " [ " + p + " ]"
+            var tables = Object.values(stream.progressSent).map(tableProgress => tableProgress.table);
+            var directionText = "To: ";
+        };
+
+        const peerWidth = {
+            width: "10%"
+        }
+        const planWidth = {
+            width: "25%"
+        }
+        const tableWidth = {
+            width: "15%"
+        }
+        const barWidth = {
+            width: "50%"
+        }
+
+        return (
+            <tr>
+                <td style={peerWidth}> <strong>{directionText} </strong> {stream.peer} </td>
+                <td style={planWidth}> <strong>PlanId: </strong> {this.props.planId} </td>
+                <td style={tableWidth}> <strong>Tables: </strong> {tables} </td>
+                <td style={barWidth}> <ProgressBar now={progress} active={isActive} label={label} bsStyle={style} key={stream.id} /> </td>
+            </tr>
+        );
+    }
+
+})
+
+export default Stream;
diff --git a/src/ui/app/jsx/streams.jsx b/src/ui/app/jsx/streams.jsx
new file mode 100644
index 000000000..a15293496
--- /dev/null
+++ b/src/ui/app/jsx/streams.jsx
@@ -0,0 +1,104 @@
+import React from "react";
+import Table from 'react-bootstrap/lib/Table';
+import Stream from 'jsx/stream';
+import {DeleteStatusMessageMixin, humanFileSize, getUrlPrefix, toast} from "jsx/mixin";
+import $ from "jquery";
+
+const Streams = React.createClass({
+
+    propTypes: {
+      endpoint: React.PropTypes.string.isRequired,
+      clusterName: React.PropTypes.string.isRequired
+    },
+
+    getInitialState() {
+      return {streamSessions: [], scheduler: {}};
+    },
+
+    componentWillMount: function() {
+      this._collectStreams();
+      this.setState({scheduler : setInterval(this._collectStreams, 10000)});
+    },
+
+    componentWillUnmount: function() {
+        clearInterval(this.state.scheduler);
+    },
+
+    _collectStreams: function() {
+      $.ajax({
+        url: getUrlPrefix(window.top.location.pathname)
+            + '/node/streams/'
+            +  encodeURIComponent(this.props.clusterName)
+            + '/'
+            + encodeURIComponent(this.props.endpoint),
+        method: 'GET',
+        component: this,
+        dataType: 'json',
+        complete: function(data) {
+            this.component.setState({streamSessions: data.responseJSON});
+        },
+        error: function(data) {
+            console.log("Failed getting streams : " + data.responseText);
+        }
+    })
+    },
+
+    _getIncomingStreams: function(streamSession) {
+        const planId = streamSession.planId
+        return Object.values(streamSession.streams)
+            .filter(stream => stream.sizeToReceive != 0)
+            .sort((s1, s2) => s1.peer > s2.peer)
+            .map(stream => <Stream planId={planId} direction="incoming" stream={stream} />)
+    },
+
+    _getOutgoingStreams: function(streamSession) {
+        const planId = streamSession.planId
+        return Object.values(streamSession.streams)
+            .filter(stream => stream.sizeToSend != 0)
+            .sort((s1, s2) => s1.peer > s2.peer)
+            .map(stream => <Stream planId={planId} direction="outgoing" stream={stream} />)
+    },
+
+    render: function() {
+
+        if (!this.state.streamSessions) {
+            console.log("No Streams found");
+            return;
+        }
+
+        const incomingStreams = this.state.streamSessions
+            .map(this._getIncomingStreams)
+            // concatenates streams from different sessions into one list
+            .reduce((sum, item) => sum.concat(item), [])
+
+        const outgoingStreams = this.state.streamSessions
+            .map(this._getOutgoingStreams)
+            // concatenates streams from different sessions into one list
+            .reduce((sum, item) => sum.concat(item), [])
+
+
+        return (
+            <div>
+                <div className="col-lg-12">
+                    <h4> Incoming Streams </h4>
+                    <Table condensed hover>
+                        <tbody>
+                            {incomingStreams}
+                        </tbody>
+                    </Table>
+                </div>
+                <div className="col-lg-12">
+                    <h4> Outgoing Streams </h4>
+                    <Table condensed hover>
+                        <tbody>
+                            {outgoingStreams}
+                        </tbody>
+                    </Table>
+                </div>
+            </div>
+        );
+    },
+
+})
+
+export default Streams;