Redisストリームを使ったリアルタイムのチャット

// track join notification
> XADD channel:welcome:log * message "adam has joined the channel."
1523499744387-0

// add to members list
> SADD channel:welcome:members "adam"
(integer) 1

// get the list of channel members
> SMEMBERS channel:welcome:members
1) "adam"

// send chat message
> XADD channel:welcome:log * message "Hi everyone!" userName "adam"
1523499778907-0

// read new messages
> XREAD BLOCK 5000 STREAMS channel:welcome:log $
1) 1) "channel:welcome:log"
   2) 1) 1) 1523499832690-0
         2) 1) "message"
            2) "Hi everyone!"
            3) "userName"
            4) "adam"
(1.67s)

// get latest 100 messages
> XREVRANGE channel:welcome:log + - COUNT 100
1) 1) 1523499832690-0
   2) 1) "message"
      2) "Hi everyone!"
      3) "userName"
      4) "adam"
2) 1) 1523499744387-0
   2) 1) "message"
      2) "adam has joined the channel."

const pollCache = {}; // storage for a shared cache of polls

function pollNewMessages (channelName) {
    if (pollCache[channelName]) return pollCache[channelName]; // use the cache

    let seenId = null; // variable to hold our position in the stream
    let key = getMessagesKey(channelName); // derive the key (i.e. channel:welcome:log)
    let connection = redis.duplicate(); // create a new connection for polling this stream

    return pollCache[channelName] = Rx.Observable.of(null) // return and cache an Observable
        .expand(() => Rx.Observable.fromPromise(connection.xread(10000, {key, id: seenId})))
        // expand calls the provided function in a loop
        // XREAD will be called on our stream key with the latest ID repeatedly
        .filter(streams => streams) // do not process empty results (i.e. time out was reached)
        .flatMap(streams => streams) // process each stream returned individually
        .flatMap(stream => stream[1]) // process each event in each stream individually
        .map(streamEvent => parseChannelMessages(streamEvent, channelName)) // parse the event
        .do(streamEvent => { // for each event
            if (streamEvent.id > seenId || !seenId) { // if it is latest seen event
                seenId = streamEvent.id; // record it as such
            }
        })
        .finally(() => { // when the stream is cleaned up
            connection.quit(); // close the redis connection
            delete pollCache[channelName]; // remove it from the cache
        })
        .publish() // wrap the observable in a shared one
        .refCount(); // track subscriptions to it so it is cleaned up automatically
}

let sockets = [];

let socketServer = socketio(server, {path: '/api/socket.io'});

Rx.Observable.fromEvent(socketServer, 'connection') // when a new connection occurs
    .flatMap(socket => { // for each connection
        sockets.push(socket); // keep list of sockets
        const part$ = Rx.Observable.fromEvent(socket, 'part'); // when user leaves a channel
        const disconnect$ = Rx.Observable.fromEvent(socket, 'disconnect') // user disconnects
            .do(() => sockets.splice(sockets.indexOf(socket), 1)); // remove socket

        return Rx.Observable.fromEvent(socket, 'join') // when user joins a channel
            .flatMap(channel => { // for the joined channel
                return data.pollNewMessages(channel) // get a poller
                    .takeUntil(part$.filter(c => c === channel)) // listen until channel is left
            })
            .do(message => socket.emit('message', message)) // emit each message on the socket
            .takeUntil(disconnect$); // stop listening if user disconnects
    })
    .finally(() => sockets.forEach((socket) => {socket.destroy()})) // destroy sockets
    .takeUntil(Rx.Observable.fromEvent(process, 'SIGINT')) // listen until server is stopped
    .subscribe(() => {}); // start processing

socket = io.connect({path: '/api/socket.io'}); // connect the web socket
message = Observable.fromEvent<IMessage>(this.socket, 'message'); // socket received messages
reconnect = Observable.fromEvent(this.socket, 'reconnect'); // socket reconnections

messages: IMessage[] = [];
latestSeenId: string;

this.channelService.message
  .filter(message => message.channel === this.name) // filter message stream to this channel
  .subscribe(message => {
    this.messages.push(message); // add to the end of the array
    if (this.channelService.current == this) { // if this channel has focus
      this.latestSeenId = message.id; // consider this message seen
    }
  });

<div class="row" *ngFor="let message of channelService.current?.messages; trackBy: trackByFn">
  <div class="col">
    <span class="text-info" *ngIf="message.userName">&lt;{{message.userName}}&gt;</span>
    <span [class.text-success]="message.join" [class.text-warning]="message.part">
      {{message.message}}
    </span>
  </div>
  <div class="col-3 col-md-3 col-lg-2 text-right">
    <span class="text-secondary mr-3 timestamp">
      {{message.id | humanRelativeDate}}
    </span>
  </div>
</div>

// join a channel
router.post('/join/:channel', async (req, res) => {
    let channel = req.params.channel;
    await data.join(channel, req.userName);
    res.send({success: true});
});

// leave a channel
router.post('/part/:channel', async (req, res) => {
    let channel = req.params.channel;
    await data.part(channel, req.userName);
    res.send({success: true});
});

// send a message to a channel
router.post('/channel/:channel', async (req, res) => {
    let channel = req.params.channel;
    let message = req.body.message;
    await data.send(channel, req.userName, message);
    res.send({success: true});
});

// get messages from a channel
router.get('/channel/:channel', async (req, res) => {
    let channel = req.params.channel;
    let before = req.query.before;
    let messages = await data.getMessages(channel, before);
    res.send({success: true, messages});
});

// get members of a channel
router.get('/channel/:channel/members', async (req, res) => {
    let channel = req.params.channel;
    let members = await data.getMembers(channel);
    res.send({success: true, members});
});

// get the user's name from the request
app.use((req, res, next) => {
    req.userName = req.get('x-username');
    next();
});

$ git clone https://github.com/aprice-/redisstreamchat
$ cd redisstreamchat
$ docker-compose up -d