concatJoin RxJS operator

Are you searching for a RxJS concat operator that works like forkJoin? Then have a look at the following simple code that will do that for you.

import { concat, EMPTY, Observable } from 'rxjs';

/**
 * Joins last values emitted by passed Observables.
 *
 * `concatJoin` is an operator that takes any number of Observables which
 * can be passed either as an array or directly as arguments. If no input
 * Observables are provided, resulting stream will complete immediately.
 *
 * `concatJoin` will wait for all passed Observables to complete one by
 * one and then it will emit an array with last values from corresponding
 * Observables. So if you pass `n` Observables to the operator, resulting
 * array will have `n` values, where first value is the last thing emitted
 * by the first Observable, second value is the last thing emitted by the
 * second Observable and so on. That means `concatJoin` will not emit more
 * than once and it will complete after that.
 *
 * @param {...Observable} sources Any number of Observables provided either
 * as an array or as an arguments passed directly to the operator.
 * @return {Observable} Observable emitting either an array of last values
 * emitted by passed Observables.
 */
export function concatJoin<T>(...sources: Array<Observable<T> | Observable<T>[]>): Observable<T[]> {
  // If the first and only other argument is an array assume it's been
  // called with `concatJoin([obs1, obs2, obs3])`
  if (sources.length === 1 && Array.isArray(sources[0])) {
    sources = sources[0] as Array<Observable<T>>;
  }

  if (sources.length === 0) {
    return EMPTY;
  }

  return new Observable((subscriber) => {
    const values: Array<T> = [];
    concat(...sources).subscribe(
      (value: T) => {
        values.push(value);
      },
      (err) => {
        subscriber.error(err);
      },
      () => {
        subscriber.next(values);
        subscriber.complete();
      }
    );
  });
}

Try out OMV5 (beta)

If you want to try out OMV5 (beta) on AMD64 compatible hardware, then simply download the ISO image and install it.

You can also use any Debian 10 image and manually install OMV5 according to this forum post.

Last but not least you can give it a try in a virtual machine using Vagrant.

$ git clone https://github.com/openmediavault/openmediavault.git
$ cd openmediavault/
$ git checkout master 
$ cd vagrant/
$ vagrant up

The following changes have been done:

  • Adapt code base to Debian 10 (Buster)
  • Use Salt to deploy configuration settings. Using ‘omv-salt stage run all’ will deploy the whole system according to the configuration database.
  • The omv-salt CLI command superseds omv-mkconf. Note, the command will not only create the configuration files, it will also take care about to start/stop/restart the services.
  • The omv-initsystem command has been replaced by ‘omv-confdbadm populate’. This command discovers the current system settings and synchronizes them into the database.
  • Default systems settings will now be applied by the Salt orchestration state ‘omv.stage.setup’.
  • Add Time Machine support to SMB/CIFS shares.
  • Use systemd-logind to handle power button action.
  • Use systemd to reboot/shutdown/standby the system.
  • Issue #93: Use chrony instead of ntpd.
  • Issue #146: Use systemd to configure the network.
  • Issue #266: Remember size and position of windows.
  • Issue #281: Display system uptime in RRD graph.
  • Issue #302: Display reboot notification.
  • Issue #317: Remove the ability to configure Zeroconf services via UI. They can still be customized via environment variables.
  • Add options ‘Group’ and ‘Owner’ to rsync settings. If the ‘Archive’ option is enabled and one of its implied option ‘Recursive’, ‘Preserve permissions’, ‘Times’, ‘Group’ or ‘Owner’ is disabled, then this option will be prefixed with ‘–no-xxx’ in the rsync command line.
  • Remove ‘Local master browser’ checkbox in the SMB/CIFS page. If you still need that use the ‘Extra Options’ field to add it manually.