channelstream: separate element and basic pubsub

frontend/src/components/channelstream.js

@@ -0,0 +1,57 b''
+// # Copyright (C) 2010-2016  RhodeCode GmbH
+// #
+// # This program is free software: you can redistribute it and/or modify
+// # it under the terms of the GNU Affero General Public License, version 3
+// # (only), as published by the Free Software Foundation.
+// #
+// # This program is distributed in the hope that it will be useful,
+// # but WITHOUT ANY WARRANTY; without even the implied warranty of
+// # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// # GNU General Public License for more details.
+// #
+// # You should have received a copy of the GNU Affero General Public License
+// # along with this program.  If not, see <http://www.gnu.org/licenses/>.
+// #
+// # This program is dual-licensed. If you wish to learn more about the
+// # AppEnlight Enterprise Edition, including its added features, Support
+// # services, and proprietary license terms, please see
+// # https://rhodecode.com/licenses/
+
+angular.module('appenlight.components.channelstream', [])
+    .component('channelstream', {
+        controller: ChannelstreamController,
+        bindings: {
+            config: '='
+        }
+    });
+
+ChannelstreamController.$inject = ['$rootScope','userSelfPropertyResource'];
+
+function ChannelstreamController($rootScope, userSelfPropertyResource){
+    userSelfPropertyResource.get({key: 'websocket'}, function (data) {
+        stateHolder.websocket = new ReconnectingWebSocket(this.config.ws_url + '/ws?conn_id=' + data.conn_id);
+        stateHolder.websocket.onopen = function (event) {
+            console.log('open');
+        };
+        stateHolder.websocket.onmessage = function (event) {
+            var data = JSON.parse(event.data);
+            _.each(data, function (message) {
+                console.log('channelstream-message', message);
+                if(typeof message.message.topic !== 'undefined'){
+                    $rootScope.$broadcast(
+                        'channelstream-message.'+message.message.topic, message);
+                }
+                else{
+                    $rootScope.$broadcast('channelstream-message', message);
+                }
+            });
+        };
+        stateHolder.websocket.onclose = function (event) {
+            console.log('closed');
+        };
+
+        stateHolder.websocket.onerror = function (event) {
+            console.log('error');
+        };
+    });
+}

backend/src/appenlight/lib/utils/__init__.py

@@ -104,7 +104,7 b' class DateTimeEncoder(json.JSONEncoder):'
         return json.JSONEncoder.default(self, obj)
 
 
-def cometd_request(secret, endpoint, payload, throw_exceptions=False,
+def channelstream_request(secret, endpoint, payload, throw_exceptions=False,
                    servers=None):
     responses = []
     if not servers:

backend/src/appenlight/models/report.py

@@ -32,7 +32,7 b' from appenlight.models import Base, Datastores'
 from appenlight.lib.utils.date_utils import convert_date
 from appenlight.lib.utils import convert_es_type
 from appenlight.models.slow_call import SlowCall
-from appenlight.lib.utils import cometd_request
+from appenlight.lib.utils import channelstream_request
 from appenlight.lib.enums import ReportType, Language
 from pyramid.threadlocal import get_current_registry, get_current_request
 from sqlalchemy.dialects.postgresql import JSON
@@ -410,7 +410,7 b' class Report(Base, BaseModel):'
         Sends notification to websocket channel
         """
         settings = get_current_registry().settings
-        log.info('notify cometd')
+        log.info('notify channelstream')
         if self.report_type != ReportType.error:
             return
         payload = {
@@ -418,7 +418,7 b' class Report(Base, BaseModel):'
             "user": '__system__',
             "channel": 'app_%s' % self.resource_id,
             'message': {
-                'type': 'report',
+                'topic': 'front_dashboard.new_topic',
                 'report': {
                     'group': {
                         'priority': report_group.priority,
@@ -441,8 +441,7 b' class Report(Base, BaseModel):'
             }
 
         }
-
-        cometd_request(settings['cometd.secret'], '/message', [payload],
+        channelstream_request(settings['cometd.secret'], '/message', [payload],
                        servers=[settings['cometd_servers']])
 
     def es_doc(self):

backend/src/appenlight/static/js/appenlight.js

@@ -2671,7 +2671,12 b" angular.module('appenlight.base', ["
 
 angular.module('appenlight.filters', []);
 angular.module('appenlight.templates', []);
-angular.module('appenlight.controllers', ['appenlight.base']);
+angular.module('appenlight.controllers', [
+    'appenlight.base'
+]);
+angular.module('appenlight.components', [
+    'appenlight.components.channelstream'
+]);
 angular.module('appenlight.directives', [
     'appenlight.directives.appVersion',
     'appenlight.directives.c3chart',
@@ -2714,6 +2719,7 b" var app = angular.module('appenlight', ["
     'appenlight.services',
     'appenlight.directives',
     'appenlight.controllers',
+    'appenlight.components',
     'appenlight.plugins'
 ]);
 
@@ -2731,7 +2737,6 b' function kickstartAE(initialUserData) {'
                     return response;
                 },
                 'responseError': function (rejection) {
-                    
                     if (rejection.status > 299 && rejection.status !== 422) {
                         stateHolder.flashMessages.extend([{
                             msg: 'Response status code: ' + rejection.status + ', "' + rejection.statusText + '", url: ' + rejection.config.url,
@@ -7017,6 +7022,64 b' function kickstartAE(initialUserData) {'
 // # services, and proprietary license terms, please see
 // # https://rhodecode.com/licenses/
 
+angular.module('appenlight.components.channelstream', [])
+    .component('channelstream', {
+        controller: ChannelstreamController,
+        bindings: {
+            config: '='
+        }
+    });
+
+ChannelstreamController.$inject = ['$rootScope','userSelfPropertyResource'];
+
+function ChannelstreamController($rootScope, userSelfPropertyResource){
+    userSelfPropertyResource.get({key: 'websocket'}, function (data) {
+        stateHolder.websocket = new ReconnectingWebSocket(this.config.ws_url + '/ws?conn_id=' + data.conn_id);
+        stateHolder.websocket.onopen = function (event) {
+            
+        };
+        stateHolder.websocket.onmessage = function (event) {
+            var data = JSON.parse(event.data);
+            _.each(data, function (message) {
+                
+                if(typeof message.message.topic !== 'undefined'){
+                    $rootScope.$broadcast(
+                        'channelstream-message.'+message.message.topic, message);
+                }
+                else{
+                    $rootScope.$broadcast('channelstream-message', message);
+                }
+            });
+        };
+        stateHolder.websocket.onclose = function (event) {
+            
+        };
+
+        stateHolder.websocket.onerror = function (event) {
+            
+        };
+    });
+}
+
+;// # Copyright (C) 2010-2016  RhodeCode GmbH
+// #
+// # This program is free software: you can redistribute it and/or modify
+// # it under the terms of the GNU Affero General Public License, version 3
+// # (only), as published by the Free Software Foundation.
+// #
+// # This program is distributed in the hope that it will be useful,
+// # but WITHOUT ANY WARRANTY; without even the implied warranty of
+// # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// # GNU General Public License for more details.
+// #
+// # You should have received a copy of the GNU Affero General Public License
+// # along with this program.  If not, see <http://www.gnu.org/licenses/>.
+// #
+// # This program is dual-licensed. If you wish to learn more about the
+// # AppEnlight Enterprise Edition, including its added features, Support
+// # services, and proprietary license terms, please see
+// # https://rhodecode.com/licenses/
+
 var aeconfig = angular.module('appenlight.config', []);
 aeconfig.factory('AeConfig', function () {
     var obj = {};
@@ -8478,21 +8541,9 b' function IndexDashboardController($scope, $location, $cookies, $interval, stateH'
         'series': true
     };
     vm.stream = {paused: false, filtered: false, messages: [], reports: []};
-    vm.websocket = null;
-    userSelfPropertyResource.get({key: 'websocket'}, function (data) {
-        
 
-        vm.websocket = new ReconnectingWebSocket(AeConfig.ws_url + '/ws?conn_id=' + data.conn_id);
-        vm.websocket.onopen = function (event) {
-            
-        };
-        vm.websocket.onmessage = function (event) {
-            var data = JSON.parse(event.data);
-            
-            $scope.$apply(function (scope) {
-                _.each(data, function (message) {
-                    if (message.type === 'message'){
-                        ws_report = message.message.report;
+    $scope.$on('channelstream-message.front_dashboard.new_topic', function(event, message){
+        var ws_report = message.message.report;
         if (ws_report.http_status != 500) {
             return
         }
@@ -8515,18 +8566,6 b' function IndexDashboardController($scope, $location, $cookies, $interval, stateH'
             }
             vm.stream.reports.unshift(ws_report);
         }
-                    }
-                });
-            });
-        };
-        vm.websocket.onclose = function (event) {
-            
-        };
-
-        vm.websocket.onerror = function (event) {
-            
-        };
-
     });
 
     vm.determineStartState = function () {
@@ -8582,12 +8621,12 b' function IndexDashboardController($scope, $location, $cookies, $interval, stateH'
         vm.fetchMetrics();
         vm.fetchRequestsBreakdown();
         vm.fetchSlowCalls();
-    }
+    };
 
     vm.changedTimeSpan = function(){
         vm.startDateFilter = timeSpanToStartDate(vm.timeSpan.key);
         vm.refreshData();
-    }
+    };
 
     var intervalId = $interval(function () {
         if (_.contains(['30m', "1h"], vm.timeSpan.key)) {
@@ -8599,14 +8638,6 b' function IndexDashboardController($scope, $location, $cookies, $interval, stateH'
         }
     }, 60000);
 
-    $scope.$on('$destroy',function(){
-        $interval.cancel(intervalId);
-        if (vm.websocket && vm.websocket.readyState == 1){
-            vm.websocket.close();
-        }
-    });
-
-
     vm.fetchApdexStats = function () {
         vm.loading.apdex = true;
         vm.apdexStats = applicationsPropertyResource.query({
@@ -12706,7 +12737,8 b" angular.module('appenlight.services.resources').factory('resourcesPropertyResour"
 // # services, and proprietary license terms, please see
 // # https://rhodecode.com/licenses/
 
-angular.module('appenlight.services.stateHolder', []).factory('stateHolder', ['$timeout', 'AeConfig', function ($timeout, AeConfig) {
+angular.module('appenlight.services.stateHolder', []).factory('stateHolder', 
+    ['$timeout', '$rootScope', 'AeConfig', function ($timeout, $rootScope, AeConfig) {
 
     var AeUser = {"user_name": null, "id": null};
     AeUser.update = function (jsonData) {

backend/src/appenlight/templates/header.jinja2

@@ -1,5 +1,6 b''
 <!-- Fixed navbar -->
 <div id="top-navbar" class="navbar navbar-default navbar-fixed-top" role="navigation"  data-ng-controller="HeaderCtrl as header">
+{% raw %}<channelstream config="AeConfig"></channelstream>{% endraw %}
 <div class="pattern">
     <div class="container">
             <div class="navbar-header pull-left">

backend/src/appenlight/views/user.py

@@ -37,7 +37,7 b' from ziggurat_foundations.models.services.external_identity import \\'
 
 from appenlight.lib import generate_random_string
 from appenlight.lib.social import handle_social_data
-from appenlight.lib.utils import cometd_request, add_cors_headers, \
+from appenlight.lib.utils import channelstream_request, add_cors_headers, \
     permission_tuple_to_dict
 from appenlight.models import DBSession
 from appenlight.models.alert_channels.email import EmailAlertChannel
@@ -259,7 +259,7 b' def users_websocket(request):'
                "channels": channels
                }
     settings = request.registry.settings
-    response = cometd_request(
+    response = channelstream_request(
         settings['cometd.secret'], '/connect', payload,
         servers=[request.registry.settings['cometd_servers']],
         throw_exceptions=True)

frontend/src/app.js

@@ -36,7 +36,12 b" angular.module('appenlight.base', ["
 
 angular.module('appenlight.filters', []);
 angular.module('appenlight.templates', []);
-angular.module('appenlight.controllers', ['appenlight.base']);
+angular.module('appenlight.controllers', [
+    'appenlight.base'
+]);
+angular.module('appenlight.components', [
+    'appenlight.components.channelstream'
+]);
 angular.module('appenlight.directives', [
     'appenlight.directives.appVersion',
     'appenlight.directives.c3chart',
@@ -79,6 +84,7 b" var app = angular.module('appenlight', ["
     'appenlight.services',
     'appenlight.directives',
     'appenlight.controllers',
+    'appenlight.components',
     'appenlight.plugins'
 ]);
 
@@ -96,7 +102,6 b' function kickstartAE(initialUserData) {'
                     return response;
                 },
                 'responseError': function (rejection) {
-                    console.log(rejection);
                     if (rejection.status > 299 && rejection.status !== 422) {
                         stateHolder.flashMessages.extend([{
                             msg: 'Response status code: ' + rejection.status + ', "' + rejection.statusText + '", url: ' + rejection.config.url,

frontend/src/controllers/front_dashboard.js

@@ -375,21 +375,9 b' function IndexDashboardController($scope, $location, $cookies, $interval, stateH'
         'series': true
     };
     vm.stream = {paused: false, filtered: false, messages: [], reports: []};
-    vm.websocket = null;
-    userSelfPropertyResource.get({key: 'websocket'}, function (data) {
-        console.log(data);
-        console.log(AeConfig.ws_url + '/ws?conn_id=' + data.conn_id);
-        vm.websocket = new ReconnectingWebSocket(AeConfig.ws_url + '/ws?conn_id=' + data.conn_id);
-        vm.websocket.onopen = function (event) {
-            console.log('open');
-        };
-        vm.websocket.onmessage = function (event) {
-            var data = JSON.parse(event.data);
-            console.log('message');
-            $scope.$apply(function (scope) {
-                _.each(data, function (message) {
-                    if (message.type === 'message'){
-                        ws_report = message.message.report;
+
+    $scope.$on('channelstream-message.front_dashboard.new_topic', function(event, message){
+        var ws_report = message.message.report;
                         if (ws_report.http_status != 500) {
                             return
                         }
@@ -412,18 +400,6 b' function IndexDashboardController($scope, $location, $cookies, $interval, stateH'
                             }
                             vm.stream.reports.unshift(ws_report);
                         }
-                    }
-                });
-            });
-        };
-        vm.websocket.onclose = function (event) {
-            console.log('closed');
-        };
-
-        vm.websocket.onerror = function (event) {
-            console.log('error');
-        };
-
     });
 
     vm.determineStartState = function () {
@@ -479,12 +455,12 b' function IndexDashboardController($scope, $location, $cookies, $interval, stateH'
         vm.fetchMetrics();
         vm.fetchRequestsBreakdown();
         vm.fetchSlowCalls();
-    }
+    };
 
     vm.changedTimeSpan = function(){
         vm.startDateFilter = timeSpanToStartDate(vm.timeSpan.key);
         vm.refreshData();
-    }
+    };
 
     var intervalId = $interval(function () {
         if (_.contains(['30m', "1h"], vm.timeSpan.key)) {
@@ -496,14 +472,6 b' function IndexDashboardController($scope, $location, $cookies, $interval, stateH'
         }
     }, 60000);
 
-    $scope.$on('$destroy',function(){
-        $interval.cancel(intervalId);
-        if (vm.websocket && vm.websocket.readyState == 1){
-            vm.websocket.close();
-        }
-    });
-
-
     vm.fetchApdexStats = function () {
         vm.loading.apdex = true;
         vm.apdexStats = applicationsPropertyResource.query({

frontend/src/services/state_holder.js

@@ -17,7 +17,8 b''
 // # services, and proprietary license terms, please see
 // # https://rhodecode.com/licenses/
 
-angular.module('appenlight.services.stateHolder', []).factory('stateHolder', ['$timeout', 'AeConfig', function ($timeout, AeConfig) {
+angular.module('appenlight.services.stateHolder', []).factory('stateHolder',
+    ['$timeout', 'AeConfig', function ($timeout, AeConfig) {
 
     var AeUser = {"user_name": null, "id": null};
     AeUser.update = function (jsonData) {