| 218 | } |
| 219 | |
| 220 | bool redis_pipeline_channel::handle_result(redis_pipeline_message* msg, |
| 221 | const redis_result* result) const |
| 222 | { |
| 223 | int type = result->get_type(); |
| 224 | if (type == REDIS_RESULT_UNKOWN) { |
| 225 | logger_warn("Unknown type=%d", (int) type); |
| 226 | msg->push(result); |
| 227 | return true; |
| 228 | } |
| 229 | if (type != REDIS_RESULT_ERROR) { |
| 230 | msg->push(result); |
| 231 | return true; |
| 232 | } |
| 233 | |
| 234 | #define EQ(x, y) !strncasecmp((x), (y), sizeof(y) -1) |
| 235 | |
| 236 | const char* ptr = result->get_error(); |
| 237 | if (ptr == NULL || *ptr == 0) { |
| 238 | logger_error("error info null"); |
| 239 | msg->push(result); |
| 240 | return true; |
| 241 | } |
| 242 | |
| 243 | if (EQ(ptr, "MOVED") || EQ(ptr, "ASK")) { |
| 244 | dbuf_pool* dbuf = msg->get_dbuf(); |
| 245 | assert(dbuf); |
| 246 | |
| 247 | const char* addr = redis_command::get_addr(dbuf, ptr); |
| 248 | if (addr == NULL) { |
| 249 | logger_error("No redirect addr got"); |
| 250 | msg->push(result); |
| 251 | } else if (msg->get_redirect_count() >= 5) { |
| 252 | logger_error("Redirect count(%zd) exceed limit(5)", |
| 253 | msg->get_redirect_count()); |
| 254 | msg->push(result); |
| 255 | } else { |
| 256 | msg->set_addr(addr); |
| 257 | msg->set_type(redis_pipeline_t_redirect); |
| 258 | |
| 259 | // Transfer the msg back to the pipeline thread again. |
| 260 | pipeline_.notify(msg); |
| 261 | } |
| 262 | return true; |
| 263 | } |
| 264 | |
| 265 | if (EQ(ptr, "CLUSTERDOWN")) { |
| 266 | // Notify the waiter the result. |
| 267 | msg->push(result); |
| 268 | |
| 269 | // And notify the pipeline thread the redis node down now, |
| 270 | // the message created here will be deleted in pipeline thread. |
| 271 | redis_pipeline_message* m = NEW redis_pipeline_message( |
| 272 | redis_pipeline_t_clusterdonw, NULL); |
| 273 | m->set_addr(this->get_addr()); |
| 274 | // Transfer the cluster down msg to the pipeline thread. |
| 275 | pipeline_.notify(m); |
| 276 | return false; // Return false to break the loop process. |
| 277 | } |