* This is called when the application transitions from one state into * another. This means that it has finished writing the data that it needed * to, or finished receiving the data that it needed to. */
| 590 | * to, or finished receiving the data that it needed to. |
| 591 | */ |
| 592 | void TNonblockingServer::TConnection::transition() { |
| 593 | // ensure this connection is active right now |
| 594 | assert(ioThread_); |
| 595 | assert(server_); |
| 596 | |
| 597 | // Switch upon the state that we are currently in and move to a new state |
| 598 | switch (appState_) { |
| 599 | |
| 600 | case APP_READ_REQUEST: |
| 601 | // We are done reading the request, package the read buffer into transport |
| 602 | // and get back some data from the dispatch function |
| 603 | if (server_->getHeaderTransport()) { |
| 604 | inputTransport_->resetBuffer(readBuffer_, readBufferPos_); |
| 605 | outputTransport_->resetBuffer(); |
| 606 | } else { |
| 607 | // We saved room for the framing size in case header transport needed it, |
| 608 | // but just skip it for the non-header case |
| 609 | inputTransport_->resetBuffer(readBuffer_ + 4, readBufferPos_ - 4); |
| 610 | outputTransport_->resetBuffer(); |
| 611 | |
| 612 | // Prepend four bytes of blank space to the buffer so we can |
| 613 | // write the frame size there later. |
| 614 | outputTransport_->getWritePtr(4); |
| 615 | outputTransport_->wroteBytes(4); |
| 616 | } |
| 617 | |
| 618 | server_->incrementActiveProcessors(); |
| 619 | |
| 620 | if (server_->isThreadPoolProcessing()) { |
| 621 | // We are setting up a Task to do this work and we will wait on it |
| 622 | |
| 623 | // Create task and dispatch to the thread manager |
| 624 | std::shared_ptr<Runnable> task = std::shared_ptr<Runnable>( |
| 625 | new Task(processor_, inputProtocol_, outputProtocol_, this)); |
| 626 | // The application is now waiting on the task to finish |
| 627 | appState_ = APP_WAIT_TASK; |
| 628 | |
| 629 | // Set this connection idle so that libevent doesn't process more |
| 630 | // data on it while we're still waiting for the threadmanager to |
| 631 | // finish this task |
| 632 | setIdle(); |
| 633 | |
| 634 | try { |
| 635 | server_->addTask(task); |
| 636 | } catch (IllegalStateException& ise) { |
| 637 | // The ThreadManager is not ready to handle any more tasks (it's probably shutting down). |
| 638 | TOutput::instance().printf("IllegalStateException: Server::process() %s", ise.what()); |
| 639 | server_->decrementActiveProcessors(); |
| 640 | close(); |
| 641 | } catch (TimedOutException& to) { |
| 642 | TOutput::instance().printf("[ERROR] TimedOutException: Server::process() %s", to.what()); |
| 643 | server_->decrementActiveProcessors(); |
| 644 | close(); |
| 645 | } |
| 646 | |
| 647 | return; |
| 648 | } else { |
| 649 | try { |
no test coverage detected