upgrade to rxjs 6

Author: joshiste

spring-boot-admin-server-ui/package-lock.json

@@ -33,7 +33,7 @@
  "popper.js": "^1.14.3",
  "pretty-bytes": "^5.1.0",
  "resize-observer-polyfill": "^1.5.0",
- "rxjs": "^5.5.11",
+ "rxjs": "^6.2.1",
  "vue": "^2.5.16",
  "vue-clickaway": "^2.2.1",
  "vue-router": "^3.0.1",

spring-boot-admin-server-ui/package.json

@@ -16,7 +16,7 @@
 
 import axios from '@/utils/axios';
 import waitForPolyfill from '@/utils/eventsource-polyfill';
-import {Observable} from '@/utils/rxjs';
+import {concat, from, ignoreElements, Observable} from '@/utils/rxjs';
 import uri from '@/utils/uri';
 import * as _ from 'lodash';
 import Instance from './instance';
@@ -46,7 +46,8 @@ class Application {
  }
 
  static getStream() {
- return Observable.from(waitForPolyfill()).ignoreElements().concat(
+ return concat(
+ from(waitForPolyfill()).pipe(ignoreElements()),
  Observable.create(observer => {
  const eventSource = new EventSource('applications');
  eventSource.onmessage = message => observer.next({
@@ -58,7 +59,8 @@ class Application {
  return () => {
  eventSource.close();
  };
- }));
+ })
+ );
  }
 
  static _transformResponse(data) {

spring-boot-admin-server-ui/src/main/frontend/services/application.js

@@ -17,7 +17,7 @@
 import axios from '@/utils/axios';
 import waitForPolyfill from '@/utils/eventsource-polyfill';
 import logtail from '@/utils/logtail';
-import {Observable} from '@/utils/rxjs'
+import {concat, from, ignoreElements, Observable} from '@/utils/rxjs';
 import uri from '@/utils/uri';
 import _ from 'lodash';
 
@@ -222,17 +222,20 @@ class Instance {
  }
 
  static getEventStream() {
- return Observable.from(waitForPolyfill()).ignoreElements().concat(Observable.create(observer => {
- const eventSource = new EventSource('instances/events');
- eventSource.onmessage = message => observer.next({
- ...message,
- data: JSON.parse(message.data)
- });
- eventSource.onerror = err => observer.error(err);
- return () => {
- eventSource.close();
- };
- }));
+ return concat(
+ from(waitForPolyfill()).pipe(ignoreElements()),
+ Observable.create(observer => {
+ const eventSource = new EventSource('instances/events');
+ eventSource.onmessage = message => observer.next({
+ ...message,
+ data: JSON.parse(message.data)
+ });
+ eventSource.onerror = err => observer.error(err);
+ return () => {
+ eventSource.close();
+ };
+ })
+ );
  }
 
  static async get(id) {

spring-boot-admin-server-ui/src/main/frontend/services/instance.js

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 import Application from '@/services/application';
-import {Observable} from '@/utils/rxjs';
+import {concat, concatMap, defer, delay, doFirst, map, retryWhen, tap} from '@/utils/rxjs';
 
 export default class {
  constructor() {
@@ -71,13 +71,19 @@ export default class {
  }
 
  start() {
- const listing = Observable.defer(() => Application.list()).concatMap(message => message.data);
- const stream = Application.getStream().map(message => message.data);
- this.subscription = listing.concat(stream)
- .doFirst(() => this._dispatchEvent('connected'))
- .retryWhen(errors => errors
- .do(error => this._dispatchEvent('error', error))
- .delay(5000)
+ const list = defer(() => Application.list())
+ .pipe(concatMap(message => message.data));
+ const stream = Application.getStream()
+ .pipe(map(message => message.data));
+ this.subscription = concat(list, stream)
+ .pipe(
+ doFirst(() => this._dispatchEvent('connected')),
+ retryWhen(
+ errors => errors.pipe(
+ tap(error => this._dispatchEvent('error', error)),
+ delay(5000)
+ )
+ )
  ).subscribe({
  next: application => {
  const idx = this.applications.indexOfApplication(application.name);

spring-boot-admin-server-ui/src/main/frontend/store.js

@@ -14,59 +14,62 @@
  * limitations under the License.
  */
 
+import {concatMap, EMPTY, of, timer} from '@/utils/rxjs';
 import axios from './axios';
-import {Observable} from './rxjs';
 
 export default (url, interval, initialSize = 300 * 1024) => {
  let range = `bytes=-${initialSize}`;
  let size = 0;
 
- return Observable.timer(0, interval)
- .concatMap(() =>
- axios.get(url, {
- headers: {
- range
- }
- }))
- .concatMap(response => {
- const initial = size === 0;
- const contentLength = response.data.length;
- if (response.status === 200) {
- if (!initial) {
- throw 'Expected 206 - Partial Content on subsequent requests.';
+ return timer(0, interval)
+ .pipe(
+ concatMap(() =>
+ axios.get(url, {
+ headers: {
+ range
+ }
+ })
+ ),
+ concatMap(response => {
+ const initial = size === 0;
+ const contentLength = response.data.length;
+ if (response.status === 200) {
+ if (!initial) {
+ throw 'Expected 206 - Partial Content on subsequent requests.';
+ }
+ size = contentLength;
+ } else if (response.status === 206) {
+ size = parseInt(response.headers['content-range'].split('/')[1]);
+ } else {
+ throw 'Unexpected response status: ' + response.status;
  }
- size = contentLength;
- } else if (response.status === 206) {
- size = parseInt(response.headers['content-range'].split('/')[1]);
- } else {
- throw 'Unexpected response status: ' + response.status;
- }
 
- // Reload the last byte to avoid a 416: Range unsatisfiable.
- // If the response has length = 1 the file hasn't beent changed.
- // If the response status is 416 the logfile has been truncated.
- range = `bytes=${size - 1}-`;
+  // Reload the last byte to avoid a 416: Range unsatisfiable.
+  // If the response has length = 1 the file hasn't beent changed.
+  // If the response status is 416 the logfile has been truncated.
+  range = `bytes=${size - 1}-`;
 
- let addendum = null;
- let skipped = 0;
+  let addendum = null;
+  let skipped = 0;
 
- if (initial) {
- if (contentLength >= size) {
- addendum = response.data;
- } else {
- // In case of a partial response find the first line break.
- addendum = response.data.substring(response.data.indexOf('\n') + 1);
- skipped = size - addendum.length;
+ if (initial) {
+ if (contentLength >= size) {
+ addendum = response.data;
+ } else {
+ // In case of a partial response find the first line break.
+ addendum = response.data.substring(response.data.indexOf('\n') + 1);
+ skipped = size - addendum.length;
+ }
+ } else if (response.data.length > 1) {
+ // Remove the first byte which has been part of the previos response.
+ addendum = response.data.substring(1);
  }
- } else if (response.data.length > 1) {
- // Remove the first byte which has been part of the previos response.
- addendum = response.data.substring(1);
- }
 
- return addendum ? Observable.of({
- totalBytes: size,
- skipped,
- addendum
- }) : Observable.empty();
- });
+ return addendum ? of({
+ totalBytes: size,
+ skipped,
+ addendum
+ }) : EMPTY;
+ })
+ );
 }

spring-boot-admin-server-ui/src/main/frontend/utils/logtail.js

@@ -14,66 +14,67 @@
  * limitations under the License.
  */
 
-import 'rxjs/add/observable/defer';
-import 'rxjs/add/observable/empty';
-import 'rxjs/add/observable/from';
-import 'rxjs/add/observable/of';
-import 'rxjs/add/observable/timer';
-import 'rxjs/add/operator/concat';
-import 'rxjs/add/operator/concatAll';
-import 'rxjs/add/operator/concatMap';
-import 'rxjs/add/operator/delay';
-import 'rxjs/add/operator/do';
-import 'rxjs/add/operator/filter';
-import 'rxjs/add/operator/ignoreElements';
-import 'rxjs/add/operator/map';
-import 'rxjs/add/operator/merge';
-import 'rxjs/add/operator/retryWhen';
+import {defer} from 'rxjs/internal/observable/defer';
+import {tap} from 'rxjs/internal/operators/tap';
 
-import {Observable} from 'rxjs/Observable';
-import {animationFrame} from 'rxjs/scheduler/animationFrame';
-import {Subject} from 'rxjs/Subject';
+export {throwError} from 'rxjs/internal/observable/throwError';
+export {of} from 'rxjs/internal/observable/of';
+export {defer} from 'rxjs/internal/observable/defer';
+export {concat} from 'rxjs/internal/observable/concat';
+export {EMPTY} from 'rxjs/internal/observable/empty';
+export {from} from 'rxjs/internal/observable/from';
+export {timer} from 'rxjs/internal/observable/timer';
+export {Observable} from 'rxjs/internal/Observable';
+export {Subject} from 'rxjs/internal/Subject';
+export {animationFrame as animationFrameScheduler} from 'rxjs/internal/scheduler/animationFrame';
 
-Observable.prototype.doOnSubscribe = function (onSubscribe) {
- let source = this;
- return Observable.defer(() => {
- onSubscribe();
- return source;
+export {concatMap} from 'rxjs/internal/operators/concatMap';
+export {delay} from 'rxjs/internal/operators/delay';
+export {merge} from 'rxjs/internal/operators/merge';
+export {map} from 'rxjs/internal/operators/map';
+export {retryWhen} from 'rxjs/internal/operators/retryWhen';
+export {tap} from 'rxjs/internal/operators/tap';
+export {filter} from 'rxjs/internal/operators/filter';
+export {concatAll} from 'rxjs/internal/operators/concatAll';
+export {ignoreElements} from 'rxjs/internal/operators/ignoreElements';
+
+
+
+export const doOnSubscribe = cb => source =>
+ defer(() => {
+ cb();
+ return source
  });
-};
 
-Observable.prototype.doFirst = function (doFirst) {
- let source = this;
- let triggered = false;
- return Observable.defer(() => {
+export const doFirst = cb => source => {
+ let triggered;
+ return defer(() => {
  triggered = false;
  return source;
- }).do(n => {
- if (!triggered) {
- triggered = true;
- doFirst(n);
- }
- });
+ }).pipe(
+ tap( v => {
+ if (!triggered) {
+ triggered = true;
+ cb(v);
+ }
+ })
+ );
 };
 
-Observable.prototype.listen = function (callbackFn) {
+export const listen = (cb, execDelay = 150) => source => {
  let handle = null;
- return this.doOnSubscribe(() => handle = setTimeout(() => callbackFn('executing'), 150))
- .do({
+ return source.pipe(
+ doOnSubscribe(() => handle = setTimeout(() => cb('executing'), execDelay)),
+ tap({
  complete: () => {
  handle && clearTimeout(handle);
- callbackFn('completed');
+ cb('completed');
  },
  error: (error) => {
  console.warn('Operation failed:', error);
  handle && clearTimeout(handle);
- callbackFn('failed');
+ cb('failed');
  }
- });
-};
-
-export {
- Observable,
- Subject,
- animationFrame
+ })
+ )
 };