This function puts a replica in the online state, and should be called just * after a replica received the RDB file for the initial synchronization, and * we are finally ready to send the incremental stream of commands. * * It does a few things: * 1) Close the replica's connection async if it doesn't need replication * commands buffer stream, since it actually isn't a valid replica. * 2)
| 1021 | * sending it to the replica. |
| 1022 | * 4) Update the count of "good replicas". */ |
| 1023 | void putSlaveOnline(client *slave) { |
| 1024 | slave->replstate = SLAVE_STATE_ONLINE; |
| 1025 | slave->repl_put_online_on_ack = 0; |
| 1026 | slave->repl_ack_time = server.unixtime; /* Prevent false timeout. */ |
| 1027 | |
| 1028 | if (slave->flags & CLIENT_REPL_RDBONLY) { |
| 1029 | serverLog(LL_NOTICE, |
| 1030 | "Close the connection with replica %s as RDB transfer is complete", |
| 1031 | replicationGetSlaveName(slave)); |
| 1032 | freeClientAsync(slave); |
| 1033 | return; |
| 1034 | } |
| 1035 | if (connSetWriteHandler(slave->conn, sendReplyToClient) == C_ERR) { |
| 1036 | serverLog(LL_WARNING,"Unable to register writable event for replica bulk transfer: %s", strerror(errno)); |
| 1037 | freeClient(slave); |
| 1038 | return; |
| 1039 | } |
| 1040 | refreshGoodSlavesCount(); |
| 1041 | /* Fire the replica change modules event. */ |
| 1042 | moduleFireServerEvent(REDISMODULE_EVENT_REPLICA_CHANGE, |
| 1043 | REDISMODULE_SUBEVENT_REPLICA_CHANGE_ONLINE, |
| 1044 | NULL); |
| 1045 | serverLog(LL_NOTICE,"Synchronization with replica %s succeeded", |
| 1046 | replicationGetSlaveName(slave)); |
| 1047 | } |
| 1048 | |
| 1049 | /* We call this function periodically to remove an RDB file that was |
| 1050 | * generated because of replication, in an instance that is otherwise |
no test coverage detected