// virtual
LLIOPipe::EStatus LLSDRPCServer::process_impl(
	const LLChannelDescriptors& channels,
	buffer_ptr_t& buffer,
	bool& eos,
	LLSD& context,
	LLPumpIO* pump)
{
	LL_RECORD_BLOCK_TIME(FTM_PROCESS_SDRPC_SERVER);
	PUMP_DEBUG;
//	LL_DEBUGS() << "LLSDRPCServer::process_impl" << LL_ENDL;
	// Once we have all the data, We need to read the sd on
	// the the in channel, and respond on  the out channel
	if(!eos) return STATUS_BREAK;
	if(!pump || !buffer) return STATUS_PRECONDITION_NOT_MET;

	std::string method_name;
	LLIOPipe::EStatus status = STATUS_DONE;

	switch(mState)
	{
	case STATE_DEFERRED:
		PUMP_DEBUG;
		if(ESDRPCS_DONE != deferredResponse(channels, buffer.get()))
		{
			buildFault(
				channels,
				buffer.get(),
				FAULT_GENERIC,
				"deferred response failed.");
		}
		mState = STATE_DONE;
		return STATUS_DONE;

	case STATE_DONE:
//		LL_DEBUGS() << "STATE_DONE" << LL_ENDL;
		break;
	case STATE_CALLBACK:
//		LL_DEBUGS() << "STATE_CALLBACK" << LL_ENDL;
		PUMP_DEBUG;
		method_name = mRequest[LLSDRPC_METHOD_SD_NAME].asString();
		if(!method_name.empty() && mRequest.has(LLSDRPC_PARAMETER_SD_NAME))
		{
			if(ESDRPCS_DONE != callbackMethod(
				   method_name,
				   mRequest[LLSDRPC_PARAMETER_SD_NAME],
				   channels,
				   buffer.get()))
			{
				buildFault(
					channels,
					buffer.get(),
					FAULT_GENERIC,
					"Callback method call failed.");
			}
		}
		else
		{
			// this should never happen, since we should not be in
			// this state unless we originally found a method and
			// params during the first call to process.
			buildFault(
				channels,
				buffer.get(),
				FAULT_GENERIC,
				"Invalid LLSDRPC sever state - callback without method.");
		}
		pump->clearLock(mLock);
		mLock = 0;
		mState = STATE_DONE;
		break;
	case STATE_NONE:
//		LL_DEBUGS() << "STATE_NONE" << LL_ENDL;
	default:
	{
		// First time we got here - process the SD request, and call
		// the method.
		PUMP_DEBUG;
		LLBufferStream istr(channels, buffer.get());
		mRequest.clear();
		LLSDSerialize::fromNotation(
			mRequest,
			istr,
			buffer->count(channels.in()));

		// { 'method':'...', 'parameter': ... }
		method_name = mRequest[LLSDRPC_METHOD_SD_NAME].asString();
		if(!method_name.empty() && mRequest.has(LLSDRPC_PARAMETER_SD_NAME))
		{
			ESDRPCSStatus rv = callMethod(
				method_name,
				mRequest[LLSDRPC_PARAMETER_SD_NAME],
				channels,
				buffer.get());
			switch(rv)
			{
			case ESDRPCS_DEFERRED:
				mPump = pump;
				mLock = pump->setLock();
				mState = STATE_DEFERRED;
//.........这里部分代码省略.........

LLIOPipe::EStatus LLHTTPPipe::process_impl(
	const LLChannelDescriptors& channels,
    buffer_ptr_t& buffer,
    bool& eos,
    LLSD& context,
    LLPumpIO* pump)
{
	PUMP_DEBUG;
    lldebugs << "LLSDHTTPServer::process_impl" << llendl;

    // Once we have all the data, We need to read the sd on
    // the the in channel, and respond on  the out channel

    if(!eos) return STATUS_BREAK;
    if(!pump || !buffer) return STATUS_PRECONDITION_NOT_MET;

	PUMP_DEBUG;
	if (mState == STATE_INVOKE)
	{
		PUMP_DEBUG;
		mState = STATE_DELAYED;
			// assume deferred unless mResponse does otherwise
		mResponse = Response::create(this);

		// *TODO: Babbage: Parameterize parser?
		// *TODO: We should look at content-type and do the right
		// thing. Phoenix 2007-12-31
		LLBufferStream istr(channels, buffer.get());

		static LLTimer timer;
		timer.reset();

		std::string verb = context[CONTEXT_REQUEST][CONTEXT_VERB];
		if(verb == HTTP_VERB_GET)
		{
            LLPerfBlock getblock("http_get");   
			mNode.get(LLHTTPNode::ResponsePtr(mResponse), context);
		}
		else if(verb == HTTP_VERB_PUT)
		{
            LLPerfBlock putblock("http_put");
			LLSD input;
			LLSDSerialize::fromXML(input, istr);
			mNode.put(LLHTTPNode::ResponsePtr(mResponse), context, input);
		}
		else if(verb == HTTP_VERB_POST)
		{
            LLPerfBlock postblock("http_post");
			LLSD input;
			LLSDSerialize::fromXML(input, istr);
			mNode.post(LLHTTPNode::ResponsePtr(mResponse), context, input);
		}
		else if(verb == HTTP_VERB_DELETE)
		{
            LLPerfBlock delblock("http_delete");
			mNode.del(LLHTTPNode::ResponsePtr(mResponse), context);
		}		
		else if(verb == HTTP_VERB_OPTIONS)
		{
			mNode.options(LLHTTPNode::ResponsePtr(mResponse), context);
		}		
		else 
		{
		    mResponse->methodNotAllowed();
		}

		F32 delta = timer.getElapsedTimeF32();
		if (sTimingCallback)
		{
			LLHTTPNode::Description desc;
			mNode.describe(desc);
			LLSD info = desc.getInfo();
			std::string timing_name = info["description"];
			timing_name += " ";
			timing_name += verb;
			sTimingCallback(timing_name.c_str(), delta, sTimingCallbackData);
		}

		// Log all HTTP transactions.
		// TODO: Add a way to log these to their own file instead of indra.log
		// It is just too spammy to be in indra.log.
		lldebugs << verb << " " << context[CONTEXT_REQUEST]["path"].asString()
			<< " " << mStatusCode << " " <<  mStatusMessage << " " << delta
			<< "s" << llendl;

		// Log Internal Server Errors
		//if(mStatusCode == 500)
		//{
		//	llwarns << "LLHTTPPipe::process_impl:500:Internal Server Error" 
		//			<< llendl;
		//}
	}

	PUMP_DEBUG;
	switch (mState)
	{
		case STATE_DELAYED:
			lockChain(pump);
			mState = STATE_LOCKED;
			return STATUS_BREAK;
//.........这里部分代码省略.........

// virtual
LLIOPipe::EStatus LLIOSocketWriter::process_impl(
	const LLChannelDescriptors& channels,
	buffer_ptr_t& buffer,
	bool& eos,
	LLSD& context,
	LLPumpIO* pump)
{
	PUMP_DEBUG;
	LLMemType m1(LLMemType::MTYPE_IO_TCP);
	if(!mDestination) return STATUS_PRECONDITION_NOT_MET;
	if(!mInitialized)
	{
		PUMP_DEBUG;
		// Since the write will not block, it's ok to initialize and
		// attempt to write immediately.
		mInitialized = true;
		if(pump)
		{
			PUMP_DEBUG;
			lldebugs << "Initializing poll descriptor for LLIOSocketWriter."
					 << llendl;
			apr_pollfd_t poll_fd;
			poll_fd.p = NULL;
			poll_fd.desc_type = APR_POLL_SOCKET;
			poll_fd.reqevents = APR_POLLOUT;
			poll_fd.rtnevents = 0x0;
			poll_fd.desc.s = mDestination->getSocket();
			poll_fd.client_data = NULL;
			pump->setConditional(this, &poll_fd);
		}
	}

	PUMP_DEBUG;
	// *FIX: Some sort of writev implementation would be much more
	// efficient - not only because writev() is better, but also
	// because we won't have to do as much work to find the start
	// address.
	LLBufferArray::segment_iterator_t it;
	LLBufferArray::segment_iterator_t end = buffer->endSegment();
	LLSegment segment;
	it = buffer->constructSegmentAfter(mLastWritten, segment);
	/*
	if(NULL == mLastWritten)
	{
		it = buffer->beginSegment();
		segment = (*it);
	}
	else
	{
		it = buffer->getSegment(mLastWritten);
		segment = (*it);
		S32 size = segment.size();
		U8* data = segment.data();
		if((data + size) == mLastWritten)
		{
			++it;
			segment = (*it);
		}
		else
		{
			// *FIX: check the math on this one
			segment = LLSegment(
				(*it).getChannelMask(),
				mLastWritten + 1,
				size - (mLastWritten - data));
		}
	}
	*/

	PUMP_DEBUG;
	apr_size_t len;
	bool done = false;
	apr_status_t status = APR_SUCCESS;
	while(it != end)
	{

		PUMP_DEBUG;
		if((*it).isOnChannel(channels.in()))
		{
			PUMP_DEBUG;
			len = (apr_size_t)segment.size();
			status = apr_socket_send(
				mDestination->getSocket(),
				(const char*)segment.data(),
				&len);
			// We sometimes get a 'non-blocking socket operation could not be 
			// completed immediately' error from apr_socket_send.  In this
			// case we break and the data will be sent the next time the chain
			// is pumped.
			if(APR_STATUS_IS_EAGAIN(status))
			{
				ll_apr_warn_status(status);
				break;
			}

			mLastWritten = segment.data() + len - 1;

			PUMP_DEBUG;
			if((S32)len < segment.size())
//.........这里部分代码省略.........

// virtual
LLIOPipe::EStatus LLURLRequest::process_impl(
	const LLChannelDescriptors& channels,
	buffer_ptr_t& buffer,
	bool& eos,
	LLSD& context,
	LLPumpIO* pump)
{
	PUMP_DEBUG;
	LLMemType m1(LLMemType::MTYPE_IO_URL_REQUEST);
	//llinfos << "LLURLRequest::process_impl()" << llendl;
	if(!buffer) return STATUS_ERROR;
	switch(mState)
	{
	case STATE_INITIALIZED:
	{
		PUMP_DEBUG;
		// We only need to wait for input if we are uploading
		// something.
		if(((HTTP_PUT == mAction) || (HTTP_POST == mAction)) && !eos)
		{
			// we're waiting to get all of the information
			return STATUS_BREAK;
		}

		// *FIX: bit of a hack, but it should work. The configure and
		// callback method expect this information to be ready.
		mDetail->mResponseBuffer = buffer.get();
		mDetail->mChannels = channels;
		if(!configure())
		{
			return STATUS_ERROR;
		}
		mState = STATE_WAITING_FOR_RESPONSE;

		// *FIX: Maybe we should just go to the next state now...
		return STATUS_BREAK;
	}
	case STATE_WAITING_FOR_RESPONSE:
	case STATE_PROCESSING_RESPONSE:
	{
		PUMP_DEBUG;
		LLIOPipe::EStatus status = STATUS_BREAK;
		mDetail->mCurlRequest->perform();
		while(1)
		{
			CURLcode result;
			bool newmsg = mDetail->mCurlRequest->getResult(&result);
			if(!newmsg)
			{
				// we're still waiting or prcessing, check how many
				// bytes we have accumulated.
				const S32 MIN_ACCUMULATION = 100000;
				if(pump && (mDetail->mByteAccumulator > MIN_ACCUMULATION))
				{
					// This is a pretty sloppy calculation, but this
					// tries to make the gross assumption that if data
					// is coming in at 56kb/s, then this transfer will
					// probably succeed. So, if we're accumlated
					// 100,000 bytes (MIN_ACCUMULATION) then let's
					// give this client another 2s to complete.
					const F32 TIMEOUT_ADJUSTMENT = 2.0f;
					mDetail->mByteAccumulator = 0;
					pump->adjustTimeoutSeconds(TIMEOUT_ADJUSTMENT);
				}

				// keep processing
				break;
			}

			mState = STATE_HAVE_RESPONSE;
			switch(result)
			{
				case CURLE_OK:
				case CURLE_WRITE_ERROR:
					// NB: The error indication means that we stopped the
					// writing due the body limit being reached
					if(mCompletionCallback && pump)
					{
						LLURLRequestComplete* complete = NULL;
						complete = (LLURLRequestComplete*)
							mCompletionCallback.get();
						complete->responseStatus(
								result == CURLE_OK
									? STATUS_OK : STATUS_STOP);
						LLPumpIO::links_t chain;
						LLPumpIO::LLLinkInfo link;
						link.mPipe = mCompletionCallback;
						link.mChannels = LLBufferArray::makeChannelConsumer(
							channels);
						chain.push_back(link);
						pump->respond(chain, buffer, context);
						mCompletionCallback = NULL;
					}
					break;
				case CURLE_FAILED_INIT:
				case CURLE_COULDNT_CONNECT:
					status = STATUS_NO_CONNECTION;
					break;
				default:
//.........这里部分代码省略.........

LLIOPipe::EStatus LLFilterXMLRPCRequest2LLSD::process_impl(
	const LLChannelDescriptors& channels,
	buffer_ptr_t& buffer,
	bool& eos,
	LLSD& context,
	LLPumpIO* pump)
{
	LL_RECORD_BLOCK_TIME(FTM_PROCESS_XMLRPC2LLSD_REQUEST);
	PUMP_DEBUG;
	if(!eos) return STATUS_BREAK;
	if(!buffer) return STATUS_ERROR;

	PUMP_DEBUG;
	// *FIX: This technique for reading data is far from optimal. We
	// need to have some kind of istream interface into the xml
	// parser...
	S32 bytes = buffer->countAfter(channels.in(), NULL);
	if(!bytes) return STATUS_ERROR;
	char* buf = new char[bytes + 1];
	buf[bytes] = '\0';
	buffer->readAfter(channels.in(), NULL, (U8*)buf, bytes);

	//LL_DEBUGS() << "xmlrpc request: " << buf << LL_ENDL;
	
	// Check the value in the buffer. XMLRPC_REQUEST_FromXML will report a error code 4 if 
	// values that are less than 0x20 are passed to it, except
	// 0x09: Horizontal tab; 0x0a: New Line; 0x0d: Carriage
	U8* cur_pBuf = (U8*)buf;
    U8 cur_char;
	for (S32 i=0; i<bytes; i++) 
	{
        cur_char = *cur_pBuf;
		if (   cur_char < 0x20
            && 0x09 != cur_char
            && 0x0a != cur_char
            && 0x0d != cur_char )
        {
			*cur_pBuf = '?';
        }
		++cur_pBuf;
	}

	PUMP_DEBUG;
	XMLRPC_REQUEST request = XMLRPC_REQUEST_FromXML(
		buf,
		bytes,
		NULL);
	if(!request)
	{
		LL_WARNS() << "XML -> SD Request process parse error." << LL_ENDL;
		delete[] buf;
		return STATUS_ERROR;
	}

	PUMP_DEBUG;
	LLBufferStream stream(channels, buffer.get());
	stream.precision(DEFAULT_PRECISION);
	const char* name = XMLRPC_RequestGetMethodName(request);
	stream << LLSDRPC_REQUEST_HEADER_1 << (name ? name : "")
		   << LLSDRPC_REQUEST_HEADER_2;
	XMLRPC_VALUE param = XMLRPC_RequestGetData(request);
	if(param)
	{
		PUMP_DEBUG;
		S32 size = XMLRPC_VectorSize(param);
		if(size > 1)
		{
			// if there are multiple parameters, stuff the values into
			// an array so that the next step in the chain can read them.
			stream << "[";
		}
 		XMLRPC_VALUE current = XMLRPC_VectorRewind(param);
		bool needs_comma = false;
 		while(current)
 		{
			if(needs_comma)
			{
				stream << ",";
			}
			needs_comma = true;
 			stream_out(stream, current);
 			current = XMLRPC_VectorNext(param);
 		}
		if(size > 1)
		{
			// close the array
			stream << "]";
		}
	}
	stream << LLSDRPC_REQUEST_FOOTER << std::flush;
	XMLRPC_RequestFree(request, 1);
	delete[] buf;
	PUMP_DEBUG;
	return STATUS_DONE;
}

// virtual
LLIOPipe::EStatus LLSDRPCClient::process_impl(
	const LLChannelDescriptors& channels,
	buffer_ptr_t& buffer,
	bool& eos,
	LLSD& context,
	LLPumpIO* pump)
{
	LLFastTimer t(FTM_PROCESS_SDRPC_CLIENT);
	PUMP_DEBUG;
	LLMemType m1(LLMemType::MTYPE_IO_SD_CLIENT);
	if((STATE_NONE == mState) || (!pump))
	{
		// You should have called the call() method already.
		return STATUS_PRECONDITION_NOT_MET;
	}
	EStatus rv = STATUS_DONE;
	switch(mState)
	{
	case STATE_READY:
	{
		PUMP_DEBUG;
//		lldebugs << "LLSDRPCClient::process_impl STATE_READY" << llendl;
		buffer->append(
			channels.out(),
			(U8*)mRequest.c_str(),
			mRequest.length());
		context[CONTEXT_DEST_URI_SD_LABEL] = mURI;
		mState = STATE_WAITING_FOR_RESPONSE;
		break;
	}
	case STATE_WAITING_FOR_RESPONSE:
	{
		PUMP_DEBUG;
		// The input channel has the sd response in it.
		//lldebugs << "LLSDRPCClient::process_impl STATE_WAITING_FOR_RESPONSE"
		//		 << llendl;
		LLBufferStream resp(channels, buffer.get());
		LLSD sd;
		LLSDSerialize::fromNotation(sd, resp, buffer->count(channels.in()));
		LLSDRPCResponse* response = (LLSDRPCResponse*)mResponse.get();
		if (!response)
		{
			mState = STATE_DONE;
			break;
		}
		response->extractResponse(sd);
		if(EPBQ_PROCESS == mQueue)
		{
			LLPumpIO::chain_t chain;
			chain.push_back(mResponse);
			pump->addChain(chain, DEFAULT_CHAIN_EXPIRY_SECS);
		}
		else
		{
			pump->respond(mResponse.get());
		}
		mState = STATE_DONE;
		break;
	}
	case STATE_DONE:
	default:
		PUMP_DEBUG;
		llinfos << "invalid state to process" << llendl;
		rv = STATUS_ERROR;
		break;
	}
	return rv;
}