EnricoPicci

import { Server } from 'http';

import { Observable } from 'rxjs';
import { Observer } from 'rxjs';

import * as socketIoServer from 'socket.io';

import {SocketObs} from './socket-obs';

EnricoPicci

import { Observable, Observer } from 'rxjs';

export class SocketObs {
    constructor(private socket: SocketIO.Socket) {}
    
    onMessageType(messageType): Observable<any> {
        return new Observable<any>((observer: Observer<any>) => {
            this.socket.on(messageType, data => observer.next(data));
        });
    }

EnricoPicci

startSocketServer(httpServer: Server) {
    sockets(httpServer, this.port).pipe(
        mergeMap(socket =>
            race(
                socket.onMessageType(MessageType.BIND_MONITOR),
                socket.onMessageType(MessageType.BIND_CONTROLLER)
            )
        )
    )
    .subscribe();

EnricoPicci

import {sockets} from './socket-io-observable';
import {SocketObs} from './socket-obs';

class MobileObjectServer {
    private mobileObjectAdded = new Subject<{mobObj: MobileObject, mobObjId: string}>();
    private mobileObjectRemoved = new Subject<string>();

    startSocketServer(httpServer: Server) {
        sockets(httpServer, this.port).pipe(
            mergeMap(socket =>

EnricoPicci

import {sockets} from './socket-io-observable';
import {SocketObs} from './socket-obs';

class MobileObjectServer {
    private mobileObjectAdded = new Subject<{mobObj: MobileObject, mobObjId: string}>();
    private mobileObjectRemoved = new Subject<string>();


    startSocketServer(httpServer: Server) {
        sockets(httpServer, this.port).pipe(

EnricoPicci

startSocketServer(httpServer: Server) {
        sockets(httpServer, this.port).pipe(
            mergeMap(socket =>
                race(
                    socket.onMessageType(MessageType.BIND_MONITOR)
                    .pipe(
                        map(() => (socketObs: SocketObs) => this.handleMonitorObs(socketObs))
                    ),
                    socket.onMessageType(MessageType.BIND_CONTROLLER)
                    .pipe(

EnricoPicci

import {sockets} from './socket-io-observable';
import {SocketObs} from './socket-obs';

class MobileObjectServer {
    private mobileObjectAdded = new Subject<{mobObj: MobileObject, mobObjId: string}>();
    private mobileObjectRemoved = new Subject<string>();


        public startSocketServer(httpServer: Server) {
        sockets(httpServer, this.port).pipe(

EnricoPicci

export const readLinesObs = Observable.bindCallback(_readLines);
function _readLines(filePath: string, callback: (lines: Array<string>) => void) {
    const lines = new Array<string>();
    const rl = readline.createInterface({
        input: fs.createReadStream(filePath),
        crlfDelay: Infinity
    });
    rl.on('line', (line: string)  => {
        lines.push(line);
    });

EnricoPicci

    type uF = (x: any) => any;
    type NUMBER = (f: uF) => (x: any) => any;    
    const TWO: NUMBER = (f: uF) => (x: any) => f(f(x));
    const f: uF = x => x + 1;
    const x = 0;
    const result = TWO(f)(x);
    console.log(result)  // 2

EnricoPicci

    const TWO: NUMBER = f => x => f(f(x));
    const f = x => x + 1;
    const x = 0;
    const result = TWO(f)(x);
    console.log(result)  // 2