/**
 * bundle socket 的时候，这个文件是 tsc 的一个 entry
 * import 一下需要的 Rx 操作符
 */ import 'rxjs/add/operator/catch' ; import 'rxjs/add/operator/toPromise' ; import 'rxjs/add/operator/concatMap' ; import 'rxjs/add/operator/take' ; import * as URL from 'url' ; import * as PATH from 'path' ; import { Observable } from 'rxjs/Observable' ; import { ReplaySubject } from 'rxjs/ReplaySubject' ; import { Net } from '../Net' ; import { Database } from '../db' ; import { SDKFetch , SDKFetchOptions } from '../SDKFetch' ; import { socketHandler , createMsgToDBHandler , createMsgHandler } from './EventMaps' ; import { Interceptors , Proxy } from './Middleware' ; import * as Consumer from 'snapper-consumer' ; import { UserMe } from '../schemas/UserMe' ; import { TableInfoByMessageType } from './MapToTable' ; import { WSMsgToDBHandler , WSMsgHandler } from '../utils' ; declare const global ; const ctx = typeof global === 'undefined' ? window : global ; const socketUrlToHostPath = ( socketUrl ) => { const postfix = '/websocket' ; const url = URL . parse ( socketUrl ) ; const host = URL . format ( { ... url , pathname : undefined } ) ; let path = url . pathname || '' ; if ( path . slice ( - postfix . length ) !== postfix ) { path = PATH . join ( path , postfix ) ; } return { host , path } ; } ; export class SocketClient { private _isDebug = false ; private _client : Consumer | undefined ; private _socketUrl = 'wss://push.teambition.com' ; private _consumerId : string | undefined ; private _getUserMeStream = new ReplaySubject < UserMe > ( 1 ) ; private _joinedRoom = new Set < string > ( ) ; private _leavedRoom = new Set < string > ( ) ; /**
     * 拦截器序列。如果需要在消息接触 db 之前对起进行额外的过滤、变换
     * 等操作，可以在这里添加相应 handler。
     */ public interceptors : Interceptors ; /**
     * 代理。如果需要获得不会接触 db 的消息（与数据模型无关），比如界面
     * 状态变更的消息等，可以在这里注册相应 handler。
     */ public proxy : Proxy ; private handleMsgToDB : WSMsgToDBHandler ; private handleMsg : WSMsgHandler ; private database : Database | undefined ; constructor ( private fetch , private net , private mapToTable ) { this . proxy = new Proxy ( ) ; this . handleMsg = createMsgHandler ( this . proxy ) ; this . interceptors = new Interceptors ( createMsgToDBHandler ( mapToTable ) ) ; this . handleMsgToDB = ( msg , db ) => { const ret = this . interceptors . apply ( msg , db ) ; return ret instanceof Observable ? ret : Observable . of ( null ) ; } ; this . net . initMsgToDBHandler ( this . handleMsgToDB ) ; } destroy ( ) { this . _getUserMeStream . complete ( ) ; } debug ( ) { this . _isDebug = true ; ctx [ 'console' ] [ 'log' ] ( 'socket debug start' ) ; } /**
     * 常见的 url 参数有（它们的值一般写在配置文件里）
     *  - `https://messaging.__domain__`（如公有云部署）
     *  - `wss://__host__/messaging`（如私有云部署）
     *
     * 注意：`url` 参数并不是最终被用来与服务器端链接的 url；目前 teambition
     * 服务端要求 websocket 链接的目标 url 以 `/websocket` 结尾，
     * 所以参数里提供的 url 如果不是以 `/websocket` 结尾（一般配置都会
     * 省略这个路径片段），请求时会在 url 的路径上加上 `websocket` 片段，
     * 如：`https://messaging.teambition.net/websocket`。
     */ setSocketUrl ( url ) { this . _socketUrl = url ; } async initClient ( client , userMe ? ) { if ( ! userMe ) { await this . _getToken ( ) ; } else { this . _getUserMeStream . next ( userMe ) ; } this . _client = client ; this . _client . _join = this . _join . bind ( this ) ; this . _client . onmessage = this . _onmessage . bind ( this ) ; this . _client . onopen = this . _onopen . bind ( this ) ; this . _getUserMeStream . subscribe ( u => { this . _client ! . getToken = ( ) => { return u . tcmToken as string ; } ; } ) ; } initReactiveDB ( database ) { this . database = database ; if ( this . _client ) { this . _client . onmessage = this . _onmessage . bind ( this ) ; } } async connect ( ) { const userMe = await this . _getUserMeStream . take ( 1 ) . toPromise ( ) ; const auth = userMe . tcmToken . split ( '.' ) [ 1 ] ; const token = JSON . parse ( window . atob ( auth ) ) ; const expire = token . exp * 1000 - 3600000 ; if ( expire < Date . now ( ) ) { await this . _getToken ( ) ; } return this . _connect ( ) ; } getConsumerId ( ) { return this . _consumerId ; } /**
     * uri 格式: :type/:id
     * eg: projects, organizations/554c83b1b2c809b4715d17b0
     */ join ( uri ) { if ( ! this . _client ) { return undefined ; } if ( this . _joinedRoom . has ( uri ) ) { return this . _client ; } return this . _client . join . call ( this . _client , uri ) ; } leave ( uri ) { if ( ! this . _consumerId ) { return Promise . reject ( new Error ( `leave room failed, no consumerId` ) ) ; } if ( this . _leavedRoom . has ( uri ) ) { return Promise . resolve ( ) ; } return this . fetch . leaveRoom ( uri , this . _consumerId ) . then ( ( ) => { if ( this . _joinedRoom ) { this . _joinedRoom . delete ( uri ) ; } this . _leavedRoom . add ( uri ) ; } ) . catch ( ( e ) => { console . error ( e ) ; } ) ; } private _onopen ( ) { this . _joinedRoom . forEach ( r => { this . fetch . joinRoom ( r , this . _consumerId ! ) ; } ) ; } private _connect ( ) { const { host , path } = socketUrlToHostPath ( this . _socketUrl ) ; return this . _getUserMeStream . take ( 1 ) . toPromise ( ) . then ( userMe => { if ( this . _client ) { this . _client . connect ( host , { path , token : userMe . tcmToken as string } ) ; } else { throw new Error ( '_client is undefined' ) ; } } ) ; } private _onmessage ( event ) { if ( this . _isDebug ) { ctx [ 'console' ] [ 'log' ] ( JSON . stringify ( event , null , 2 ) ) ; } return socketHandler ( this . net , event , this . handleMsgToDB , this . handleMsg , this . mapToTable , this . database ) . toPromise ( ) . then ( null , ( err ) => ctx [ 'console' ] [ 'error' ] ( err ) ) ; } private _getToken ( ) { return this . fetch . getUserMe ( ) . toPromise ( ) . then ( ( r ) => { this . _getUserMeStream . next ( r ) ; } ) ; } private _join ( room , consumerId ) { this . _consumerId = consumerId ; return this . fetch . joinRoom ( room , consumerId ) . then ( ( ) => { this . _joinedRoom . add ( room ) ; } ) . catch ( e => { console . error ( e ) ; } ) ; } } export function leaveRoom ( this , room , consumerId , options ? ) { return this . delete < void > ( `${ room }/subscribe` , { consumerId } , options ) . toPromise ( ) ; } export function joinRoom ( this , room , consumerId , options ? ) { return this . post < void > ( `${ room }/subscribe` , { consumerId } , options ) . toPromise ( ) ; } SDKFetch . prototype . leaveRoom = leaveRoom ; SDKFetch . prototype . joinRoom = joinRoom ; declare module '../SDKFetch' { interface SDKFetch { joinRoom : typeof joinRoom ; leaveRoom : typeof leaveRoom ; } }