diff --git a/configs/development.ini b/configs/development.ini index f9bb25cb..c3d1a72b 100644 --- a/configs/development.ini +++ b/configs/development.ini @@ -667,12 +667,13 @@ search.location = %(here)s/data/index channelstream.enabled = true ; server address for channelstream server on the backend -channelstream.server = channelstream:9800 +channelstream.server = channelstream:8000 ; location of the channelstream server from outside world ; use ws:// for http or wss:// for https. This address needs to be handled ; by external HTTP server such as Nginx or Apache ; see Nginx/Apache configuration examples in our docs +; For development, comment this out to use auto-generated proxy URL channelstream.ws_url = ws://rhodecode.yourserver.com/_channelstream channelstream.secret = ENV_GENERATED channelstream.history.location = /var/opt/rhodecode_data/channelstream_history diff --git a/configs/production.ini b/configs/production.ini index 5beb696c..defb2512 100644 --- a/configs/production.ini +++ b/configs/production.ini @@ -629,7 +629,7 @@ search.location = %(here)s/data/index channelstream.enabled = true ; server address for channelstream server on the backend -channelstream.server = channelstream:9800 +channelstream.server = channelstream:8000 ; location of the channelstream server from outside world ; use ws:// for http or wss:// for https. This address needs to be handled diff --git a/rhodecode/public/js/src/components/channelstream-connection/channelstream-connection.js b/rhodecode/public/js/src/components/channelstream-connection/channelstream-connection.js index 1d665f40..152dc928 100644 --- a/rhodecode/public/js/src/components/channelstream-connection/channelstream-connection.js +++ b/rhodecode/public/js/src/components/channelstream-connection/channelstream-connection.js @@ -1,577 +1,440 @@ -import {Polymer, html} from '@polymer/polymer/polymer-legacy'; -import '@polymer/iron-ajax/iron-ajax.js'; +import {LitElement} from 'lit'; -const elemTemplate = html` - +/** + * Simplified Lit version of channelstream-connection + * Replaces iron-ajax with native fetch() + */ +export class ChannelstreamConnection extends LitElement { + static properties = { + channels: {type: Array}, + username: {type: String}, + connectionId: {type: String}, + websocket: {type: Object}, + websocketUrl: {type: String}, + connectUrl: {type: String}, + disconnectUrl: {type: String}, + subscribeUrl: {type: String}, + unsubscribeUrl: {type: String}, + messageUrl: {type: String}, + longPollUrl: {type: String}, + shouldReconnect: {type: Boolean}, + heartbeats: {type: Boolean}, + increaseBounceIv: {type: Number}, + _currentBounceIv: {type: Number}, + useWebsocket: {type: Boolean}, + connected: {type: Boolean}, + channelsState: {type: Object}, + userState: {type: Object} + }; - + constructor() { + super(); + this.channels = []; + this.username = 'Anonymous'; + this.connectionId = ''; + this.websocket = null; + this.websocketUrl = ''; + this.connectUrl = ''; + this.disconnectUrl = ''; + this.subscribeUrl = ''; + this.unsubscribeUrl = ''; + this.messageUrl = ''; + this.longPollUrl = ''; + this.shouldReconnect = true; + this.heartbeats = true; + this.increaseBounceIv = 2000; + this._currentBounceIv = 0; + this.useWebsocket = true; + this.connected = false; + this.channelsState = {}; + this.userState = {}; - + // Mutators - simplified to just arrays + this.mutators = { + connect: [], + message: [], + subscribe: [], + unsubscribe: [], + disconnect: [] + }; - + this._heartbeatInterval = null; + } - + // No render needed - this component has no template + render() { + return null; + } - -` - -Polymer({ - is: 'channelstream-connection', - - _template: elemTemplate, - - /** - * Fired when `channels` array changes. - * - * @event channelstream-channels-changed - */ - - /** - * Fired when `connect()` method succeeds. - * - * @event channelstream-connected - */ - - /** - * Fired when `connect` fails. - * - * @event channelstream-connect-error - */ - - /** - * Fired when `disconnect()` succeeds. - * - * @event channelstream-disconnected - */ - - /** - * Fired when `message()` succeeds. - * - * @event channelstream-message-sent - */ - - /** - * Fired when `message()` fails. - * - * @event channelstream-message-error - */ - - /** - * Fired when `subscribe()` succeeds. - * - * @event channelstream-subscribed - */ - - /** - * Fired when `subscribe()` fails. - * - * @event channelstream-subscribe-error - */ - - /** - * Fired when `unsubscribe()` succeeds. - * - * @event channelstream-unsubscribed - */ - - /** - * Fired when `unsubscribe()` fails. - * - * @event channelstream-unsubscribe-error - */ - - /** - * Fired when listening connection receives a message. - * - * @event channelstream-listen-message - */ - - /** - * Fired when listening connection is opened. - * - * @event channelstream-listen-opened - */ - - /** - * Fired when listening connection is closed. - * - * @event channelstream-listen-closed - */ - - /** - * Fired when listening connection suffers an error. - * - * @event channelstream-listen-error - */ - - properties: { - isReady: Boolean, - /** List of channels user should be subscribed to. */ - channels: { - type: Array, - value: function () { - return [] - }, - notify: true - }, - /** Username of connecting user. */ - username: { - type: String, - value: 'Anonymous', - reflectToAttribute: true - }, - /** Connection identifier. */ - connectionId: { - type: String, - reflectToAttribute: true - }, - /** Websocket instance. */ - websocket: { - type: Object, - value: null - }, - /** Websocket connection url. */ - websocketUrl: { - type: String, - value: '' - }, - /** URL used in `connect()`. */ - connectUrl: { - type: String, - value: '' - }, - /** URL used in `disconnect()`. */ - disconnectUrl: { - type: String, - value: '' - }, - /** URL used in `subscribe()`. */ - subscribeUrl: { - type: String, - value: '' - }, - /** URL used in `unsubscribe()`. */ - unsubscribeUrl: { - type: String, - value: '' - }, - /** URL used in `message()`. */ - messageUrl: { - type: String, - value: '' - }, - /** Long-polling connection url. */ - longPollUrl: { - type: String, - value: '' - }, - /** Long-polling connection url. */ - shouldReconnect: { - type: Boolean, - value: true - }, - /** Should send heartbeats. */ - heartbeats: { - type: Boolean, - value: true - }, - /** How much should every retry interval increase (in milliseconds) */ - increaseBounceIv: { - type: Number, - value: 2000 - }, - _currentBounceIv: { - type: Number, - reflectToAttribute: true, - value: 0 - }, - /** Should use websockets or long-polling by default */ - useWebsocket: { - type: Boolean, - reflectToAttribute: true, - value: true - }, - connected: { - type: Boolean, - reflectToAttribute: true, - value: false + updated(changedProperties) { + if (changedProperties.has('channels')) { + this.dispatchEvent(new CustomEvent('channelstream-channels-changed', { + detail: {value: this.channels} + })); } - }, + } - observers: [ - '_handleChannelsChange(channels.splices)' - ], - - listeners: { - 'channelstream-connected': 'startListening', - 'channelstream-connect-error': 'retryConnection', - }, + connectedCallback() { + super.connectedCallback(); + this.dispatchEvent(new CustomEvent('start-listening', {detail: {}})); + } /** - * Mutators hold functions that you can set locally to change the data - * that the client is sending to all endpoints - * you can call it like `elem.mutators('connect', yourFunc())` - * mutators will be executed in order they were pushed onto arrays - * + * Helper method for fetch requests */ - mutators: { - connect: function () { - return [] - }(), - message: function () { - return [] - }(), - subscribe: function () { - return [] - }(), - unsubscribe: function () { - return [] - }(), - disconnect: function () { - return [] - }() - }, - ready: function () { - this.isReady = true; - }, + async _fetchJSON(url, options = {}) { + const {body, headers, ...fetchOptions} = options; + + const response = await fetch(url, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + ...headers + }, + body: body ? JSON.stringify(body) : undefined, + ...fetchOptions + }); + + if (!response.ok) { + throw new Error(`HTTP ${response.status}: ${response.statusText}`); + } + + return response.json(); + } /** - * Connects user and fetches connection id from the server. - * + * Connects user and fetches connection id from the server */ - connect: function () { - var request = this.$['ajaxConnect']; - request.url = this.connectUrl; - request.body = { + async connect() { + const body = { username: this.username, channels: this.channels }; - for (var i = 0; i < this.mutators.connect.length; i++) { - this.mutators.connect[i](request); + + // Apply mutators + for (let mutator of this.mutators.connect) { + mutator({body}); } - request.generateRequest() - }, + + try { + const response = await this._fetchJSON(this.connectUrl, {body}); + this._handleConnect({detail: {response}}); + } catch (error) { + this._handleConnectError({detail: {error}}); + } + } + /** - * Overwrite with custom function that will + * Subscribes user to channels */ - addMutator: function (type, func) { - this.mutators[type].push(func); - }, - /** - * Subscribes user to channels. - * - */ - subscribe: function (channels) { - var request = this.$['ajaxSubscribe']; - request.url = this.subscribeUrl; - request.body = { + async subscribe(channels) { + const body = { channels: channels, conn_id: this.connectionId }; - for (var i = 0; i < this.mutators.subscribe.length; i++) { - this.mutators.subscribe[i](request); - } - if (request.body.channels.length) { - request.generateRequest(); - } - }, - /** - * Unsubscribes user from channels. - * - */ - unsubscribe: function (unsubscribe) { - var request = this.$['ajaxUnsubscribe']; - request.url = this.unsubscribeUrl; - request.body = { - channels: unsubscribe, + // Apply mutators + for (let mutator of this.mutators.subscribe) { + mutator({body}); + } + + if (body.channels.length) { + try { + const response = await this._fetchJSON(this.subscribeUrl, {body}); + this._handleSubscribe({detail: {response}}); + } catch (error) { + this.dispatchEvent(new CustomEvent('channelstream-subscribe-error', {detail: error})); + } + } + } + + /** + * Unsubscribes user from channels + */ + async unsubscribe(channels) { + const body = { + channels: channels, conn_id: this.connectionId }; - for (var i = 0; i < this.mutators.unsubscribe.length; i++) { - this.mutators.unsubscribe[i](request); + + // Apply mutators + for (let mutator of this.mutators.unsubscribe) { + mutator({body}); } - request.generateRequest() - }, + + try { + const response = await this._fetchJSON(this.unsubscribeUrl, {body}); + this._handleUnsubscribe({detail: {response}}); + } catch (error) { + this.dispatchEvent(new CustomEvent('channelstream-unsubscribe-error', {detail: error})); + } + } /** - * calculates list of channels we should add user to based on difference - * between channels property and passed channel list + * Calculates list of channels to add user to */ - calculateSubscribe: function (channels) { - var currentlySubscribed = this.channels; - var toSubscribe = []; - for (var i = 0; i < channels.length; i++) { + calculateSubscribe(channels) { + const currentlySubscribed = this.channels; + const toSubscribe = []; + for (let i = 0; i < channels.length; i++) { if (currentlySubscribed.indexOf(channels[i]) === -1) { toSubscribe.push(channels[i]); } } - return toSubscribe - }, + return toSubscribe; + } + /** - * calculates list of channels we should remove user from based difference - * between channels property and passed channel list + * Calculates list of channels to remove user from */ - calculateUnsubscribe: function (channels) { - var currentlySubscribed = this.channels; - var toUnsubscribe = []; - for (var i = 0; i < channels.length; i++) { + calculateUnsubscribe(channels) { + const currentlySubscribed = this.channels; + const toUnsubscribe = []; + for (let i = 0; i < channels.length; i++) { if (currentlySubscribed.indexOf(channels[i]) !== -1) { toUnsubscribe.push(channels[i]); } } - return toUnsubscribe - }, - /** - * Marks the connection as expired. - * - */ - disconnect: function () { - var request = this.$['ajaxDisconnect']; - request.url = this.disconnectUrl; - request.params = { - conn_id: this.connectionId - }; - for (var i = 0; i < this.mutators.disconnect.length; i++) { - this.mutators.disconnect[i](request); - } - // mark connection as expired - request.generateRequest(); - // disconnect existing connection - this.closeConnection(); - }, + return toUnsubscribe; + } /** - * Sends a message to the server. - * + * Marks the connection as expired */ - message: function (message) { - var request = this.$['ajaxMessage']; - request.url = this.messageUrl; - request.body = message; - for (var i = 0; i < this.mutators.message.length; i++) { - this.mutators.message[i](request) + async disconnect() { + const params = new URLSearchParams({conn_id: this.connectionId}); + + // Apply mutators + for (let mutator of this.mutators.disconnect) { + mutator({params}); } - request.generateRequest(); - }, + + try { + const response = await fetch(`${this.disconnectUrl}?${params}`); + const data = await response.json(); + this._handleDisconnect({detail: {response: data}}); + } catch (error) { + console.error('Disconnect error:', error); + } + + // Disconnect existing connection + this.closeConnection(); + } + /** - * Opens "long lived" (websocket/longpoll) connection to the channelstream server. - * + * Sends a message to the server */ - startListening: function (event) { - this.fire('start-listening', {}); + async message(messageData) { + const body = messageData; + + // Apply mutators + for (let mutator of this.mutators.message) { + mutator({body}); + } + + try { + const response = await this._fetchJSON(this.messageUrl, {body}); + this._handleMessage({detail: {response}}); + } catch (error) { + this._handleMessageError({detail: {error}}); + } + } + + /** + * Opens long lived connection (websocket/longpoll) + */ + startListening(event) { + this.dispatchEvent(new CustomEvent('start-listening', {detail: {}})); + if (this.useWebsocket) { this.useWebsocket = window.WebSocket ? true : false; } + if (this.useWebsocket) { this.openWebsocket(); - } - else { + } else { this.openLongPoll(); } - }, + } + /** - * Opens websocket connection. - * + * Opens websocket connection */ - openWebsocket: function () { - var url = this.websocketUrl + '?conn_id=' + this.connectionId; + openWebsocket() { + const url = this.websocketUrl + '?conn_id=' + this.connectionId; this.websocket = new WebSocket(url); this.websocket.onopen = this._handleListenOpen.bind(this); this.websocket.onclose = this._handleListenCloseEvent.bind(this); this.websocket.onerror = this._handleListenErrorEvent.bind(this); this.websocket.onmessage = this._handleListenMessageEvent.bind(this); - }, + } + /** - * Opens long-poll connection. - * + * Opens long-poll connection */ - openLongPoll: function () { - var request = this.$['ajaxListen']; - request.url = this.longPollUrl + '?conn_id=' + this.connectionId; - request.generateRequest() - }, + async openLongPoll() { + const url = this.longPollUrl + '?conn_id=' + this.connectionId; + + try { + this._handleListenOpen(); + const response = await fetch(url); + const text = await response.text(); + this._handleListenMessageEvent({data: text}); + } catch (error) { + this._handleListenError({detail: {error}}); + } + } + /** - * Retries `connect()` call while incrementing interval between tries up to 1 minute. - * + * Retries connect() call with increasing interval */ - retryConnection: function () { + retryConnection() { if (!this.shouldReconnect) { return; } + if (this._currentBounceIv < 60000) { this._currentBounceIv = this._currentBounceIv + this.increaseBounceIv; - } - else { + } else { this._currentBounceIv = 60000; } + setTimeout(this.connect.bind(this), this._currentBounceIv); - }, + } + /** - * Closes listening connection. - * + * Closes listening connection */ - closeConnection: function () { - var request = this.$['ajaxListen']; + closeConnection() { if (this.websocket && this.websocket.readyState === WebSocket.OPEN) { this.websocket.onclose = null; this.websocket.onerror = null; this.websocket.close(); } - if (request.loading) { - request.lastRequest.abort(); + + if (this._heartbeatInterval) { + clearInterval(this._heartbeatInterval); + this._heartbeatInterval = null; } + this.connected = false; - }, + } - _handleChannelsChange: function (event) { - // do not fire the event if set() didn't mutate anything - // is this a reliable way to do it? - if (!this.isReady || event === undefined) { - return + /** + * Add mutator function + */ + addMutator(type, func) { + this.mutators[type].push(func); + } + + /** + * Create heartbeat interval + */ + createHeartBeats() { + if (!this._heartbeatInterval && this.websocket !== null && this.heartbeats) { + this._heartbeatInterval = setInterval(this._sendHeartBeat.bind(this), 10000); } - this.fire('channelstream-channels-changed', event) - }, + } - _handleListenOpen: function (event) { - this.connected = true; - this.fire('channelstream-listen-opened', event); - this.createHeartBeats(); - }, - - createHeartBeats: function () { - if (typeof self._heartbeat === 'undefined' && this.websocket !== null - && this.heartbeats) { - self._heartbeat = setInterval(this._sendHeartBeat.bind(this), 10000); - } - }, - - _sendHeartBeat: function () { - if (this.websocket.readyState === WebSocket.OPEN && this.heartbeats) { + /** + * Send heartbeat + */ + _sendHeartBeat() { + if (this.websocket && this.websocket.readyState === WebSocket.OPEN && this.heartbeats) { this.websocket.send(JSON.stringify({type: 'heartbeat'})); } - }, - - _handleListenError: function (event) { - this.connected = false; - this.retryConnection(); - }, - _handleConnectError: function (event) { - this.connected = false; - this.fire('channelstream-connect-error', event.detail); - }, - - _handleListenMessageEvent: function (event) { - var data = null; - // comes from iron-ajax - if (event.detail) { - data = JSON.parse(event.detail.response) - // comes from websocket - setTimeout(this.openLongPoll.bind(this), 0); - } else { - data = JSON.parse(event.data) - } - this.fire('channelstream-listen-message', data); - - }, - - _handleListenCloseEvent: function (event) { - this.connected = false; - this.fire('channelstream-listen-closed', event.detail); - this.retryConnection(); - }, - - _handleListenErrorEvent: function (event) { - this.connected = false; - this.fire('channelstream-listen-error', {}) - }, - - _handleConnect: function (event) { - this.currentBounceIv = 0; - this.connectionId = event.detail.response.conn_id; - this.fire('channelstream-connected', event.detail.response); - }, - - _handleDisconnect: function (event) { - this.connected = false; - this.fire('channelstream-disconnected', {}); - }, - - _handleMessage: function (event) { - this.fire('channelstream-message-sent', event.detail.response); - }, - _handleMessageError: function (event) { - this.fire('channelstream-message-error', event.detail); - }, - - _handleSubscribe: function (event) { - this.fire('channelstream-subscribed', event.detail.response); - }, - - _handleSubscribeError: function (event) { - this.fire('channelstream-subscribe-error', event.detail); - }, - - _handleUnsubscribe: function (event) { - this.fire('channelstream-unsubscribed', event.detail.response); - }, - - _handleUnsubscribeError: function (event) { - this.fire('channelstream-unsubscribe-error', event.detail); } -}); + + // Event handlers + _handleListenOpen(event) { + this.connected = true; + this.dispatchEvent(new CustomEvent('channelstream-listen-opened', {detail: event})); + this.createHeartBeats(); + } + + _handleListenError(event) { + this.connected = false; + this.retryConnection(); + } + + _handleConnectError(event) { + this.connected = false; + this.dispatchEvent(new CustomEvent('channelstream-connect-error', {detail: event.detail})); + } + + _handleListenMessageEvent(event) { + let data = null; + + // From long-poll (fetch) + if (event.detail && event.detail.response) { + data = JSON.parse(event.detail.response); + setTimeout(this.openLongPoll.bind(this), 0); + } + // From websocket + else if (event.data) { + data = JSON.parse(event.data); + } + + if (data) { + this.dispatchEvent(new CustomEvent('channelstream-listen-message', {detail: data})); + } + } + + _handleListenCloseEvent(event) { + this.connected = false; + this.dispatchEvent(new CustomEvent('channelstream-listen-closed', {detail: event})); + this.retryConnection(); + } + + _handleListenErrorEvent(event) { + this.connected = false; + this.dispatchEvent(new CustomEvent('channelstream-listen-error', {detail: {}})); + } + + _handleConnect(event) { + this._currentBounceIv = 0; + this.connectionId = event.detail.response.conn_id; + this.dispatchEvent(new CustomEvent('channelstream-connected', {detail: event.detail.response})); + } + + _handleDisconnect(event) { + this.connected = false; + this.dispatchEvent(new CustomEvent('channelstream-disconnected', {detail: {}})); + } + + _handleMessage(event) { + this.dispatchEvent(new CustomEvent('channelstream-message-sent', {detail: event.detail.response})); + } + + _handleMessageError(event) { + this.dispatchEvent(new CustomEvent('channelstream-message-error', {detail: event.detail})); + } + + _handleSubscribe(event) { + this.dispatchEvent(new CustomEvent('channelstream-subscribed', {detail: event.detail.response})); + } + + _handleUnsubscribe(event) { + this.dispatchEvent(new CustomEvent('channelstream-unsubscribed', {detail: event.detail.response})); + } + + // Polymer compatibility methods - keep these for rhodecode-app + push(path, value) { + if (path === 'channels') { + this.channels = [...this.channels, value]; + } + } + + set(path, value) { + if (typeof path === 'string') { + this[path] = value; + } else if (Array.isArray(path)) { + // Handle nested path like ['channelsState', key] + if (path.length === 2 && path[0] === 'channelsState') { + this.channelsState = {...this.channelsState, [path[1]]: value}; + } + } + } +} + +customElements.define('channelstream-connection', ChannelstreamConnection); diff --git a/rhodecode/public/js/src/components/rhodecode-app/rhodecode-app.js b/rhodecode/public/js/src/components/rhodecode-app/rhodecode-app.js index 27f9b53d..5b5e6fac 100644 --- a/rhodecode/public/js/src/components/rhodecode-app/rhodecode-app.js +++ b/rhodecode/public/js/src/components/rhodecode-app/rhodecode-app.js @@ -108,6 +108,8 @@ export class RhodecodeApp extends LitElement { channelstreamConnection.websocketUrl = CHANNELSTREAM_URLS.ws + "/ws"; channelstreamConnection.longPollUrl = CHANNELSTREAM_URLS.longpoll + "/listen"; + channelstreamConnection.username = + window.templateContext.rhodecode_user.username; // some channels might already be registered by topic for (var i = 0; i < channels.length; i++) { channelstreamConnection.push("channels", channels[i]); diff --git a/rhodecode/tests/rhodecode.ini b/rhodecode/tests/rhodecode.ini index c628eba7..7a8726cb 100644 --- a/rhodecode/tests/rhodecode.ini +++ b/rhodecode/tests/rhodecode.ini @@ -588,7 +588,7 @@ search.location = %(here)s/.rc-test-data/data/index channelstream.enabled = false ; server address for channelstream server on the backend -channelstream.server = channelstream:9800 +channelstream.server = channelstream:8000 ; location of the channelstream server from outside world ; use ws:// for http or wss:// for https. This address needs to be handled