import { isLuaTable , ServerTickFunctions } from "../internal" ; import MessagingService from "./MessagingService" ; const Players = game . GetService ( "Players" ) ; const IS_STUDIO = game . GetService ( "RunService" ) . IsStudio ( ) ; interface IQueuedMessage { Name : string ; Data : unknown ; } export interface ISubscriptionMessage { Data : unknown ; Sent : number ; } interface IJobData { JobId : string ; InnerData : unknown ; } export interface ISubscriptionJobIdMessage extends ISubscriptionMessage { Data : IJobData ; } param value export function isSubscriptionMessage ( value ) { if ( isLuaTable ( value ) ) { const hasData = value . has ( "Data" ) ; return hasData ; } else { return false ; } } function isJobTargetMessage ( value ) { if ( isSubscriptionMessage ( value ) ) { if ( isLuaTable ( value . Data ) ) { return value . Data . has ( "jobId" ) ; } } return false ; } const globalMessageQueue = new Array < IQueuedMessage > ( ) ; let lastQueueTick = 0 ; let globalEventMessageCounter = 0 ; let globalSubscriptionCounter = 0 ; function processMessageQueue ( ) { if ( tick ( ) >= lastQueueTick + 60 ) { globalEventMessageCounter = 0 ; globalSubscriptionCounter = 0 ; lastQueueTick = tick ( ) ; while ( globalMessageQueue . size ( ) > 0 ) { const message = globalMessageQueue . pop ( ) ! ; MessagingService . PublishAsync ( message . Name , message . Data ) ; globalEventMessageCounter ++ ; } if ( globalEventMessageCounter >= MessagingEvent . GetMessageLimit ( ) ) { warn ( "[rbx-net] Too many messages are being sent, any further messages will be queued!" ) ; } } } /**
 * Message Size: 1kB
 * MessagesPerMin: 150 + 60 * NUMPLAYERS
 * MessagesPerTopicMin: 30M
 * MessagesPerUniversePerMin: 30M
 * SubsPerServer: 5 + 2 * numPlayers
 * SubsPerUniverse: 10K
 */ type JobIdMessage < TMessage > = { jobId : string ; message : TMessage ; } ; see https export default class MessagingEvent < TMessage extends unknown = unknown > { constructor ( private name ) { } /**
     * Gets the message limit
     */ public static GetMessageLimit ( ) { return 150 + 60 * Players . GetPlayers ( ) . size ( ) ; } /**
     * Gets the subscription limit
     */ public static GetSubscriptionLimit ( ) { return 5 + 2 * Players . GetPlayers ( ) . size ( ) ; } param data private sendToAllServersOrQueue ( data ) { const limit = MessagingEvent . GetMessageLimit ( ) ; if ( globalEventMessageCounter >= limit ) { warn ( `[rbx-net] Exceeded message limit of ${ limit }, adding to queue...` ) ; globalMessageQueue . push ( { Name : this . name , Data : data } ) ; } else { globalEventMessageCounter ++ ; MessagingService . PublishAsync ( this . name , data ) ; } } param jobId param message public SendToServer ( jobId , message ) { this . sendToAllServersOrQueue ( { jobId , message } ) ; } param message public SendToAllServers ( message ) { this . sendToAllServersOrQueue ( message ) ; } param handler public Connect ( handler ) { const limit = MessagingEvent . GetSubscriptionLimit ( ) ; if ( globalSubscriptionCounter >= limit ) { error ( `[rbx-net] Exceeded Subscription limit of ${ limit }!` ) ; } globalSubscriptionCounter ++ ; return MessagingService . SubscribeAsync ( this . name , ( data , sent ) => { const recieved = { Data : data , Sent : sent } ; const { Sent } = recieved ; if ( isJobTargetMessage ( recieved ) ) { const { Data } = recieved ; if ( game . JobId === Data . JobId ) { handler ( Data . InnerData as TMessage , Sent ) ; } } else { handler ( recieved . Data as TMessage , Sent ) ; } } ) ; } } ServerTickFunctions . push ( processMessageQueue ) ;