scripts/autoscale.js
function handleMessages(provider, bigIp, options) {
        const deferred = q.defer();
        const instanceIdsBeingAdded = [];
        const actions = [];
        const actionPromises = [];

        let messageMetadata = [];

        if (this.instance.isMaster && !options.blockSync) {
            actions.push(CloudProvider.MESSAGE_ADD_TO_CLUSTER);
        }

        if (!this.instance.isMaster) {
            actions.push(CloudProvider.MESSAGE_SYNC_COMPLETE);
        }

        provider.getMessages(actions, { toInstanceId: this.instanceId })
            .then((messages) => {
                const readPromises = [];
                const messagesArray = messages ? messages.slice() : [];

                logger.debug('Handling', messages.length, 'message(s)');

                messagesArray.forEach((message) => {
                    messageMetadata.push(
                        {
                            action: message.action,
                            toInstanceId: message.toInstanceId,
                            fromInstanceId: message.fromInstanceId
                        }
                    );
                    readPromises.push(readMessageData.call(this, provider, bigIp, message.data));
                });

                logger.silly('number of messages to read:', readPromises.length);

                return q.all(readPromises);
            })
            .then((readMessages) => {
                let metadata;
                let messageData;

                const alreadyAdding = function (instanceId) {
                    return instanceIdsBeingAdded.find((element) => {
                        return instanceId === element.toInstanceId;
                    });
                };

                const readMessagesArray = readMessages ? readMessages.slice() : [];

                logger.silly('number of read messages:', readMessagesArray.length);

                for (let i = 0; i < readMessagesArray.length; ++i) {
                    metadata = messageMetadata[i];
                    logger.silly('metadata:', metadata);

                    try {
                        messageData = JSON.parse(readMessagesArray[i]);
                    } catch (err) {
                        logger.warn('JSON.parse error:', err);
                        messageData = undefined;
                        deferred.reject(new Error('Unable to JSON parse message'));
                    }

                    if (messageData) {
                        let discard = false;
                        switch (metadata.action) {
                        // Add an instance to our cluster
                        case CloudProvider.MESSAGE_ADD_TO_CLUSTER:
                            logger.silly('message MESSAGE_ADD_TO_CLUSTER');

                            if (alreadyAdding(metadata.fromInstanceId)) {
                                logger.debug('Already adding', metadata.fromInstanceId, ', discarding');
                                discard = true;
                            }

                            if (!discard) {
                                instanceIdsBeingAdded.push({
                                    toInstanceId: metadata.fromInstanceId,
                                    fromUser: bigIp.user,
                                    fromPassword: bigIp.password
                                });

                                actionPromises.push(
                                    bigIp.cluster.joinCluster(
                                        messageData.deviceGroup,
                                        messageData.host,
                                        messageData.username,
                                        messageData.password,
                                        true,
                                        {
                                            remotePort: messageData.port,
                                            remoteHostname: messageData.hostname,
                                            passwordEncrypted: false
                                        }
                                    )
                                );
                            }

                            break;

                        // sync is complete
                        case CloudProvider.MESSAGE_SYNC_COMPLETE:
                            logger.silly('message MESSAGE_SYNC_COMPLETE');
                            actionPromises.push(
                                provider.syncComplete(messageData.fromUser, messageData.fromPassword)
                            );

                            break;
                        default:
                            logger.warn('Unknown message action', metadata.action);
                        }
                    }
                }

                return q.all(actionPromises);
            })
            .then((responses) => {
                const messagePromises = [];
                let messageData;

                const responsesArray = responses ? responses.slice() : [];

                if (instanceIdsBeingAdded.length > 0) {
                    messageMetadata = [];

                    logger.silly('responses from join cluster', responsesArray);
                    for (let i = 0; i < responsesArray.length; i++) {
                        // responsesArray[i] === true iff that instance was successfully synced
                        if (responsesArray[i] === true) {
                            logger.silly(
                                'sync is complete for instance',
                                instanceIdsBeingAdded[i].toInstanceId
                            );

                            messageMetadata.push(
                                {
                                    action: CloudProvider.MESSAGE_SYNC_COMPLETE,
                                    toInstanceId: instanceIdsBeingAdded[i].toInstanceId,
                                    fromInstanceId: this.instanceId
                                }
                            );
                            messageData = {
                                fromUser: instanceIdsBeingAdded[i].fromUser,
                                fromPassword: instanceIdsBeingAdded[i].fromPassword
                            };

                            messagePromises.push(
                                prepareMessageData.call(
                                    this,
                                    provider,
                                    instanceIdsBeingAdded[i].toInstanceId,
                                    JSON.stringify(messageData)
                                )
                            );
                        }
                    }
                }

                return q.all(messagePromises);
            })
            .then((preppedMessageData) => {
                const syncCompletePromises = [];
                let metadata;
                let messageData;

                for (let i = 0; i < preppedMessageData.length; i++) {
                    metadata = messageMetadata[i];
                    messageData = preppedMessageData[i];

                    syncCompletePromises.push(
                        provider.sendMessage(
                            CloudProvider.MESSAGE_SYNC_COMPLETE,
                            {
                                toInstanceId: metadata.toInstanceId,
                                fromInstanceId: metadata.fromInstanceId,
                                data: messageData
                            }
                        )
                    );
                }

                return q.all(syncCompletePromises);
            })
            .then(() => {
                deferred.resolve();
            })
            .catch((err) => {
                logger.warn('Error handling messages', err);
                deferred.reject(err);
            });

        return deferred.promise;
    }
function handleMessages(provider, bigIp, options) {
        const deferred = q.defer();
        const instanceIdsBeingAdded = [];
        const actions = [];
        const actionPromises = [];

        let messageMetadata = [];

        if (this.instance.isMaster && !options.blockSync) {
            actions.push(CloudProvider.MESSAGE_ADD_TO_CLUSTER);
        }

        if (!this.instance.isMaster) {
            actions.push(CloudProvider.MESSAGE_SYNC_COMPLETE);
        }

        provider.getMessages(actions, { toInstanceId: this.instanceId })
            .then((messages) => {
                const readPromises = [];
                const messagesArray = messages ? messages.slice() : [];

                logger.debug('Handling', messages.length, 'message(s)');

                messagesArray.forEach((message) => {
                    messageMetadata.push(
                        {
                            action: message.action,
                            toInstanceId: message.toInstanceId,
                            fromInstanceId: message.fromInstanceId
                        }
                    );
                    readPromises.push(readMessageData.call(this, provider, bigIp, message.data));
                });

                logger.silly('number of messages to read:', readPromises.length);

                return q.all(readPromises);
            })
            .then((readMessages) => {
                let metadata;
                let messageData;

                const alreadyAdding = function (instanceId) {
                    return instanceIdsBeingAdded.find((element) => {
                        return instanceId === element.toInstanceId;
                    });
                };

                const readMessagesArray = readMessages ? readMessages.slice() : [];

                logger.silly('number of read messages:', readMessagesArray.length);

                for (let i = 0; i < readMessagesArray.length; ++i) {
                    metadata = messageMetadata[i];
                    logger.silly('metadata:', metadata);

                    try {
                        messageData = JSON.parse(readMessagesArray[i]);
                    } catch (err) {
                        logger.warn('JSON.parse error:', err);
                        messageData = undefined;
                        deferred.reject(new Error('Unable to JSON parse message'));
                    }

                    if (messageData) {
                        let discard = false;
                        switch (metadata.action) {
                        // Add an instance to our cluster
                        case CloudProvider.MESSAGE_ADD_TO_CLUSTER:
                            logger.silly('message MESSAGE_ADD_TO_CLUSTER');

                            if (alreadyAdding(metadata.fromInstanceId)) {
                                logger.debug('Already adding', metadata.fromInstanceId, ', discarding');
                                discard = true;
                            }

                            if (!discard) {
                                instanceIdsBeingAdded.push({
                                    toInstanceId: metadata.fromInstanceId,
                                    fromUser: bigIp.user,
                                    fromPassword: bigIp.password
                                });

                                actionPromises.push(
                                    bigIp.cluster.joinCluster(
                                        messageData.deviceGroup,
                                        messageData.host,
                                        messageData.username,
                                        messageData.password,
                                        true,
                                        {
                                            remotePort: messageData.port,
                                            remoteHostname: messageData.hostname,
                                            passwordEncrypted: false
                                        }
                                    )
                                );
                            }

                            break;

                        // sync is complete
                        case CloudProvider.MESSAGE_SYNC_COMPLETE:
                            logger.silly('message MESSAGE_SYNC_COMPLETE');
                            actionPromises.push(
                                provider.syncComplete(messageData.fromUser, messageData.fromPassword)
                            );

                            break;
                        default:
                            logger.warn('Unknown message action', metadata.action);
                        }
                    }
                }

                return q.all(actionPromises);
            })
            .then((responses) => {
                const messagePromises = [];
                let messageData;

                const responsesArray = responses ? responses.slice() : [];

                if (instanceIdsBeingAdded.length > 0) {
                    messageMetadata = [];

                    logger.silly('responses from join cluster', responsesArray);
                    for (let i = 0; i < responsesArray.length; i++) {
                        // responsesArray[i] === true iff that instance was successfully synced
                        if (responsesArray[i] === true) {
                            logger.silly(
                                'sync is complete for instance',
                                instanceIdsBeingAdded[i].toInstanceId
                            );

                            messageMetadata.push(
                                {
                                    action: CloudProvider.MESSAGE_SYNC_COMPLETE,
                                    toInstanceId: instanceIdsBeingAdded[i].toInstanceId,
                                    fromInstanceId: this.instanceId
                                }
                            );
                            messageData = {
                                fromUser: instanceIdsBeingAdded[i].fromUser,
                                fromPassword: instanceIdsBeingAdded[i].fromPassword
                            };

                            messagePromises.push(
                                prepareMessageData.call(
                                    this,
                                    provider,
                                    instanceIdsBeingAdded[i].toInstanceId,
                                    JSON.stringify(messageData)
                                )
                            );
                        }
                    }
                }

                return q.all(messagePromises);
            })
            .then((preppedMessageData) => {
                const syncCompletePromises = [];
                let metadata;
                let messageData;

                for (let i = 0; i < preppedMessageData.length; i++) {
                    metadata = messageMetadata[i];
                    messageData = preppedMessageData[i];

                    syncCompletePromises.push(
                        provider.sendMessage(
                            CloudProvider.MESSAGE_SYNC_COMPLETE,
                            {
                                toInstanceId: metadata.toInstanceId,
                                fromInstanceId: metadata.fromInstanceId,
                                data: messageData
                            }
                        )
                    );
                }

                return q.all(syncCompletePromises);
            })
            .then(() => {
                deferred.resolve();
            })
            .catch((err) => {
                logger.warn('Error handling messages', err);
                deferred.reject(err);
            });

        return deferred.promise;
    }
function handleMessages ( provider , bigIp , options ) { const deferred = q . defer ( ) ; const instanceIdsBeingAdded = [ ] ; const actions = [ ] ; const actionPromises = [ ] ; let messageMetadata = [ ] ; if ( this . instance . isMaster && ! options . blockSync ) { actions . push ( CloudProvider . MESSAGE_ADD_TO_CLUSTER ) ; } if ( ! this . instance . isMaster ) { actions . push ( CloudProvider . MESSAGE_SYNC_COMPLETE ) ; } provider . getMessages ( actions , { toInstanceId : this . instanceId } ) . then ( ( messages ) => { const readPromises = [ ] ; const messagesArray = messages ? messages . slice ( ) : [ ] ; logger . debug ( 'Handling' , messages . length , 'message(s)' ) ; messagesArray . forEach ( ( message ) => { messageMetadata . push ( { action : message . action , toInstanceId : message . toInstanceId , fromInstanceId : message . fromInstanceId } ) ; readPromises . push ( readMessageData . call ( this , provider , bigIp , message . data ) ) ; } ) ; logger . silly ( 'number of messages to read:' , readPromises . length ) ; return q . all ( readPromises ) ; } ) . then ( ( readMessages ) => { let metadata ; let messageData ; const alreadyAdding = function ( instanceId ) { return instanceIdsBeingAdded . find ( ( element ) => { return instanceId === element . toInstanceId ; } ) ; } ; const readMessagesArray = readMessages ? readMessages . slice ( ) : [ ] ; logger . silly ( 'number of read messages:' , readMessagesArray . length ) ; for ( let i = 0 ; i < readMessagesArray . length ; ++ i ) { metadata = messageMetadata [ i ] ; logger . silly ( 'metadata:' , metadata ) ; try { messageData = JSON . parse ( readMessagesArray [ i ] ) ; } catch ( err ) { logger . warn ( 'JSON.parse error:' , err ) ; messageData = undefined ; deferred . reject ( new Error ( 'Unable to JSON parse message' ) ) ; } if ( messageData ) { let discard = false ; switch ( metadata . action ) { // Add an instance to our cluster case CloudProvider . MESSAGE_ADD_TO_CLUSTER : logger . silly ( 'message MESSAGE_ADD_TO_CLUSTER' ) ; if ( alreadyAdding ( metadata . fromInstanceId ) ) { logger . debug ( 'Already adding' , metadata . fromInstanceId , ', discarding' ) ; discard = true ; } if ( ! discard ) { instanceIdsBeingAdded . push ( { toInstanceId : metadata . fromInstanceId , fromUser : bigIp . user , fromPassword : bigIp . password } ) ; actionPromises . push ( bigIp . cluster . joinCluster ( messageData . deviceGroup , messageData . host , messageData . username , messageData . password , true , { remotePort : messageData . port , remoteHostname : messageData . hostname , passwordEncrypted : false } ) ) ; } break ; // sync is complete case CloudProvider . MESSAGE_SYNC_COMPLETE : logger . silly ( 'message MESSAGE_SYNC_COMPLETE' ) ; actionPromises . push ( provider . syncComplete ( messageData . fromUser , messageData . fromPassword ) ) ; break ; default : logger . warn ( 'Unknown message action' , metadata . action ) ; } } } return q . all ( actionPromises ) ; } ) . then ( ( responses ) => { const messagePromises = [ ] ; let messageData ; const responsesArray = responses ? responses . slice ( ) : [ ] ; if ( instanceIdsBeingAdded . length > 0 ) { messageMetadata = [ ] ; logger . silly ( 'responses from join cluster' , responsesArray ) ; for ( let i = 0 ; i < responsesArray . length ; i ++ ) { // responsesArray[i] === true iff that instance was successfully synced if ( responsesArray [ i ] === true ) { logger . silly ( 'sync is complete for instance' , instanceIdsBeingAdded [ i ] . toInstanceId ) ; messageMetadata . push ( { action : CloudProvider . MESSAGE_SYNC_COMPLETE , toInstanceId : instanceIdsBeingAdded [ i ] . toInstanceId , fromInstanceId : this . instanceId } ) ; messageData = { fromUser : instanceIdsBeingAdded [ i ] . fromUser , fromPassword : instanceIdsBeingAdded [ i ] . fromPassword } ; messagePromises . push ( prepareMessageData . call ( this , provider , instanceIdsBeingAdded [ i ] . toInstanceId , JSON . stringify ( messageData ) ) ) ; } } } return q . all ( messagePromises ) ; } ) . then ( ( preppedMessageData ) => { const syncCompletePromises = [ ] ; let metadata ; let messageData ; for ( let i = 0 ; i < preppedMessageData . length ; i ++ ) { metadata = messageMetadata [ i ] ; messageData = preppedMessageData [ i ] ; syncCompletePromises . push ( provider . sendMessage ( CloudProvider . MESSAGE_SYNC_COMPLETE , { toInstanceId : metadata . toInstanceId , fromInstanceId : metadata . fromInstanceId , data : messageData } ) ) ; } return q . all ( syncCompletePromises ) ; } ) . then ( ( ) => { deferred . resolve ( ) ; } ) . catch ( ( err ) => { logger . warn ( 'Error handling messages' , err ) ; deferred . reject ( err ) ; } ) ; return deferred . promise ; }
Called with this bound to the caller
Called with this bound to the caller