work with rxjs

[richardo2016]

use orm with rxjs.

import coroutine = require('coroutine')
import ORM = require('@fxjs/orm')
import KV = require('fib-kv')
import Pool = require('fib-pool')

import { interval } from 'rxjs';

let synced = false;
const ormPool = Pool<FxOrmNS.ORM>({
    create: () => {
        const orm = ORM.connectSync(CONN) as FxOrmNS.ORM;

        ;(() => {
            require('./defs/role')(orm);
            require('./defs/user')(orm);

            if (!synced) {
                const sync_lock = new coroutine.Lock();
                sync_lock.acquire();
                
                orm.dropSync();
                orm.syncSync();
                
                synced = true;
                sync_lock.release();
            }

            orm.models.user.afterSave(function (success: boolean) {
                if (!success)
                    return ;

                const roles: any[] = this.getRolesSync() || []
                kvs.userRoleIds.set(this.id, JSON.stringify(roles.map(role => role.id)))

                console.log(
                    'kvs.userRoleIds',
                    kvs.userRoleIds.get(this.id)
                )
            }, { oldhook: 'prepend' });

            orm.models.role.afterSave(function (success: boolean) {
                if (!success)
                    return ;

                const users: any[] = this.getUsersSync() || []
                kvs.roleUserIds.set(this.id, JSON.stringify(users.map(user => user.id)))

                console.log(
                    'kvs.roleUserIds',
                    kvs.roleUserIds.get(this.id)
                )
            }, { oldhook: 'prepend' });
        })()

        return orm;
    },
    destroy: (orm: FxOrmNS.ORM) => {
        orm.closeSync()
    }
})

const CONN = 'sqlite:test.db'
const kvs = {
    userRoleIds: new KV(ormPool(orm => orm.driver.db.conn), {table_name: 'cache_user_roleids'}),
    roleUserIds: new KV(ormPool(orm => orm.driver.db.conn), {table_name: 'cache_role_userids'}),
}

Object.values(kvs).forEach(kv => kv.setup())

const USERID = 1

process.nextTick(() => {
    // Fiber 1: fetch per 1s
    const interval$ = interval(1000);

    interval$.subscribe((val: any) => {
        ormPool(orm => {
            const user = orm.models.user.findSync()[0]
            if (user)
                user.roles = user.getRolesSync()

            // console.log('data', user)
        })
    });
})

process.nextTick(() => {
    // Fiber 2: fake update
    let user: any = null
    const getUser = (orm: FxOrmNS.ORM, userId = USERID) => {
        if (!user) {
            user = orm.models.user.createSync({
                username: `testuser${userId}`,
                email: `testuser${userId}@gmail.com`
            })
        }

        return user;
    }

    const getRole = (orm: FxOrmNS.ORM, roleName: string = '') => {
        let role = orm.models.role.findSync({name: roleName})[0]
        if (!role)
            role = orm.models.role.createSync({name: roleName})

        return role
    }
    
    let timeCount = 1
    let roleName = ''

    interval(200).subscribe(() => timeCount++)
    while (true) {
        // update role per 30s
        roleName = `role${timeCount}`

        ormPool(orm => {
            orm.trans(() => {
                const user = getUser(orm, USERID)

                const nextRole = getRole(orm, roleName)

                if (!user.hasRolesSync(nextRole))
                    user.addRolesSync(nextRole)

                const mainRole = getRole(orm, 'mainRole')

                if (!user.hasMainRoleSync(mainRole))
                    user.setMainRoleSync(mainRole)
                else
                    user.removeMainRoleSync(mainRole)
            })
        })
    }
})