1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172

use futures_util::StreamExt;
use log::{debug, error};
use std::sync::RwLock;
use tokio;
use trezm_redis::AsyncCommands;
use trezm_redis::RedisResult;

use tokio::sync::broadcast::channel as unbounded;
use tokio::sync::broadcast::Sender;

use crate::rooms::get_sockets_for_room;
use crate::sid::generate_sid;
use crate::socketio::{InternalMessage, SocketIOAdapter};
use crate::socketio_message::SocketIOMessage;

lazy_static! {
    static ref CHANNEL: RwLock<Vec<Sender<SocketIOToRedisMessage>>> = RwLock::new(Vec::new());
}

#[derive(Clone)]
pub struct RedisAdapter {}

impl SocketIOAdapter for RedisAdapter {
    fn incoming(&self, room_id: &str, message: &SocketIOMessage) {
        // Here we need to relay the message to the redis pubsub
        // This is client -> us -> redis
        send_message(room_id, message.clone())
    }

    fn outgoing(&self, _room_id: &str, _message: &SocketIOMessage) {
        // Here we need to forward the message
        // This is redis -> us -> client
        // Automagically handled by the listener
    }
}

#[derive(Clone)]
struct SocketIOToRedisMessage {
    room_id: String,
    socket_io_message: SocketIOMessage,
}

#[derive(Serialize, Deserialize, Debug)]
struct RedisMessage {
    channel: String,
    room_id: String,
    event: String,
    message: String,
    sending_id: String,
}

pub fn send_message(room_id: &str, message: SocketIOMessage) {
    for sender in &*CHANNEL.read().unwrap() {
        let socket_io_to_redis_message = match message {
            SocketIOMessage::Message(ref event, ref message) => Some(SocketIOToRedisMessage {
                room_id: room_id.to_owned(),
                socket_io_message: SocketIOMessage::Message(event.clone(), message.clone()),
            }),
            SocketIOMessage::SendMessage(ref event, ref message) => Some(SocketIOToRedisMessage {
                room_id: room_id.to_owned(),
                socket_io_message: SocketIOMessage::SendMessage(event.clone(), message.clone()),
            }),
            SocketIOMessage::Join(_) => None,
            SocketIOMessage::AddListener(_, _) => None,
            _ => {
                error!(
                    "Received a message that was not RawMessage, Message, or SendMessage: {}",
                    message
                );
                None
            }
        };

        if let Some(val) = socket_io_to_redis_message {
            let _ = sender.send(val);
        }
    }
}

pub async fn connect_to_pubsub(redis_host: &str, channel_name: &str) -> RedisResult<()> {
    let redis_host = redis_host.to_string();
    let channel_name = channel_name.to_string();

    let client = trezm_redis::Client::open(redis_host).unwrap();
    let mut publish_conn = client.get_async_connection().await?;

    let (sender, mut receiver) = unbounded(16);

    CHANNEL.write().unwrap().push(sender);

    let channel_name = channel_name.to_string();
    let channel_name_outgoing = channel_name.clone();
    let channel_name_incoming = channel_name.clone();
    let sending_id = generate_sid();
    let sending_id_outgoing = sending_id.clone();
    let sending_id_incoming = sending_id.clone();

    // Handle pubbing local requests into redis
    tokio::spawn(async move {
        while let Ok(val) = receiver.recv().await {
            debug!("local -> redis: {} {}", val.room_id, val.socket_io_message);

            match val.socket_io_message {
                SocketIOMessage::SendMessage(event, message) => {
                    let _ = publish_conn
                        .publish::<'_, _, _, String>(
                            channel_name_outgoing.clone(),
                            serde_json::to_string(&RedisMessage {
                                channel: channel_name_outgoing.clone(),
                                room_id: val.room_id.clone(),
                                event,
                                message,
                                sending_id: sending_id_outgoing.clone(),
                            })
                            .unwrap(),
                        )
                        .await;
                }
                SocketIOMessage::Message(event, message) => {
                    let _ = publish_conn
                        .publish::<'_, _, _, String>(
                            channel_name_outgoing.clone(),
                            serde_json::to_string(&RedisMessage {
                                channel: channel_name_outgoing.clone(),
                                room_id: val.room_id.clone(),
                                event,
                                message,
                                sending_id: sending_id_outgoing.clone(),
                            })
                            .unwrap(),
                        )
                        .await;
                }
                _ => (),
            }
        }
    });

    // Handle subbing local requests from redis
    tokio::spawn(async move {
        let mut pubsub_conn = client.get_async_connection().await.unwrap().into_pubsub();

        pubsub_conn.subscribe(channel_name_incoming).await.unwrap();
        let mut pubsub_stream = pubsub_conn.on_message();

        while let Some(msg) = pubsub_stream.next().await {
            let message: RedisMessage =
                serde_json::from_str(&msg.get_payload::<String>().unwrap()).unwrap();

            debug!(
                "redis -> local: {} {} {}",
                message.room_id, message.event, message.message
            );

            if message.sending_id != sending_id_incoming {
                match get_sockets_for_room(&message.room_id) {
                    Some(sockets) => {
                        for socket in &*sockets {
                            socket.send(InternalMessage::IO(SocketIOMessage::SendMessage(
                                message.event.to_string(),
                                message.message.to_string(),
                            )));
                        }
                    }
                    None => (),
                };
            }
        }
    });

    Ok(())
}