Source Code (Use browser search to find items of interest.)

kdelibs'Connection (./kdelibs/kio/connection.h:46)

    class Connection : public QObject
    {
	Q_OBJECT
    public:
	Connection();
	virtual ~Connection();
	
	void init(KSocket *sock);
	void connect(QObject *receiver = 0, const char *member = 0);
	void close();
	
	int fd_from() const { return fd_in; }
	
	void init(int fd_in, int fd_out);
	
	bool inited() const { return (fd_in != 0) && (f_out != 0); }
	
	// send (queues the command to be sent)
	void send(int cmd, const QByteArray &arr = QByteArray());

	// send (without queue)
	bool sendnow( int _cmd, const QByteArray &data );

	/**
	 * Receive data
	 *
	 * @return >=0 indicates the received data size upon success
	 *         -1  indicates error
	 */
	int read( int* _cmd, QByteArray & );

        /**
         * Don't handle incoming data until resumed
         */
        void suspend();

        /**
         * Resume handling of incoming data
         */
        void resume();

	void queueOnly(bool queue);

        /**
         * @internal
         */
        static void sigpipe_handler(int);

    protected slots:
        void dequeue();
	
    protected:
	
	
    private:
	bool queueonly;
	int fd_in;
	FILE *f_out;
	KSocket *socket;
	QSocketNotifier *notifier;
	QObject *receiver;
	const char *member;
	QList<Task> tasks;
	int unqueuedTasks;
        bool m_suspended;
    };

};

kdelibs'Connection::Connection() (./kdelibs/arts/mcop/connection.cc:27)

Connection::Connection() :_refCnt(1)
{
	_connState = unknown;
}

kdelibs'Connection::~Connection() (./kdelibs/arts/mcop/connection.cc:32)

Connection::~Connection()
{
	assert(_refCnt == 0);
}

kdelibs'Connection::_copy() (./kdelibs/arts/mcop/connection.cc:37)

void Connection::_copy()
{
	_refCnt++;
}

kdelibs'Connection::_release() (./kdelibs/arts/mcop/connection.cc:42)

void Connection::_release()
{
	assert(_refCnt > 0);
	_refCnt--;
	if(_refCnt == 0)
		delete this;
}

kdelibs'Connection::initReceive() (./kdelibs/arts/mcop/connection.cc:50)

void Connection::initReceive()
{
	rcbuf = 0;
	receiveHeader = true;
	remaining = 12;
}

kdelibs'Connection::receive() (./kdelibs/arts/mcop/connection.cc:57)

void Connection::receive(unsigned char *data, long len)
{
	if(len > remaining)
	{
		unsigned char *data2 = data+remaining;
		long len2 = len-remaining;
		receive(data,remaining);

		/* This could be optimized to a non recursive thing (fixme?) */
		receive(data2,len2);
		return;
	}
	// get a buffer for the incoming message:
	if(!rcbuf) rcbuf = new Buffer;

	remaining -= len;
	rcbuf->write(data,len);

#ifdef DEBUG_IO
	printf("read %ld bytes\n",len);
#endif

	if(remaining == 0)
	{
		if(receiveHeader)
		{
			long mcopMagic;

			mcopMagic = rcbuf->readLong();
			remaining = rcbuf->readLong() - 12;
			messageType = rcbuf->readLong();

			if(_connState != Connection::established && remaining >= 4096)
			{
				/*
				 * don't accept large amounts of data on unauthenticated
				 * connections
				 */
				remaining = 0;
			}

			if(mcopMagic == MCOP_MAGIC)
			{
				// do we need to receive more data (message body?)
				if(remaining)
				{
					receiveHeader = false;
				}
				else
				{
					Buffer *received = rcbuf;
					initReceive();
					Dispatcher::the()->handle(this,received,messageType);
				}
			}
			else
			{
				initReceive();
				Dispatcher::the()->handleCorrupt(this);
			}
		}
		else
		{
			Buffer *received = rcbuf;

			/*
			 * it's important to prepare to receive new messages *before*
			 * calling Dispatcher::the()->handle(...), as handle may
			 * get into an I/O situation (e.g. when doing an invocation
			 * itself), and we may receive more messages while handle is
			 * running
			 */
			initReceive();

			// rcbuf is consumed by the dispatcher
			Dispatcher::the()->handle(this,received,messageType);
		}
	}
}

kdelibs'Connection::Connection() (./kdelibs/kio/connection.cpp:51)

Connection::Connection()
{
    signal( SIGPIPE, sigpipe_handler );
    f_out = 0;
    fd_in = 0;
    socket = 0;
    notifier = 0;
    receiver = 0;
    member = 0;
    queueonly = false;
    m_suspended = false;
    unqueuedTasks = 0;
}

kdelibs'Connection::~Connection() (./kdelibs/kio/connection.cpp:65)

Connection::~Connection()
{
    close();
}

kdelibs'Connection::suspend() (./kdelibs/kio/connection.cpp:70)

void Connection::suspend()
{
    m_suspended = true;
    if (notifier)
       notifier->setEnabled(false);
}

kdelibs'Connection::resume() (./kdelibs/kio/connection.cpp:77)

void Connection::resume()
{
    m_suspended = false;
    if (notifier)
       notifier->setEnabled(true);
}

kdelibs'Connection::close() (./kdelibs/kio/connection.cpp:84)

void Connection::close()
{
    delete notifier;
    notifier = 0;
    delete socket;
    socket = 0;
    if (f_out)
       fclose(f_out);
    f_out = 0;
    fd_in = 0;
}

kdelibs'Connection::send() (./kdelibs/kio/connection.cpp:96)

void Connection::send(int cmd, const QByteArray& data)
{
    if (!inited() || queueonly || tasks.count() > 0) {
	kdDebug(7017) << "pending queue " << cmd << endl;
	Task *task = new Task();
	task->cmd = cmd;
	task->data = data;
	tasks.append(task);
    } else {
	sendnow( cmd, data );
    }

    if (inited() && tasks.count() > 0 && !queueonly)
	QTimer::singleShot(100, this, SLOT(dequeue()));
}

kdelibs'Connection::queueOnly() (./kdelibs/kio/connection.cpp:112)

void Connection::queueOnly(bool queue) {
    unqueuedTasks = tasks.count();
    kdDebug(7017) << "setting queueOnly to " << queue << endl;
    queueonly = queue;
    dequeue();
}

kdelibs'Connection::dequeue() (./kdelibs/kio/connection.cpp:119)

void Connection::dequeue()
{
    if (tasks.count() == 0  || !inited() || (queueonly && unqueuedTasks == 0))
	return;

    kdDebug(7017) << "dequeue" << endl;

    tasks.first();
    Task *task = tasks.take();
    sendnow( task->cmd, task->data );
    delete task;

    if (queueonly)
	unqueuedTasks--;

    if (tasks.count() > 0 && (!queueonly || unqueuedTasks > 0))
	QTimer::singleShot(100, this, SLOT(dequeue()));
}

kdelibs'Connection::init() (./kdelibs/kio/connection.cpp:139)

void Connection::init(int _fd_in, int fd_out)
{
    fd_in = _fd_in;
    f_out = fdopen( fd_out, "wb" );
}

kdelibs'Connection::init() (./kdelibs/kio/connection.cpp:145)

void Connection::init(KSocket *sock)
{
    delete notifier;
    notifier = 0;
    delete socket;
    socket = sock;
    fd_in = socket->socket();
    f_out = fdopen( socket->socket(), "wb" );
    if (receiver && fd_in) {
	notifier = new QSocketNotifier(fd_in, QSocketNotifier::Read);
        if ( m_suspended )
            suspend();
	QObject::connect(notifier, SIGNAL(activated(int)), receiver, member);
    }
    dequeue();
}

kdelibs'Connection::connect() (./kdelibs/kio/connection.cpp:162)

void Connection::connect(QObject *_receiver, const char *_member)
{
    receiver = _receiver;
    member = _member;
    delete notifier;
    notifier = 0;
    if (receiver && fd_in) {
	notifier = new QSocketNotifier(fd_in, QSocketNotifier::Read);
        if ( m_suspended )
            suspend();
	QObject::connect(notifier, SIGNAL(activated(int)), receiver, member);
    }
}

kdelibs'Connection::sendnow() (./kdelibs/kio/connection.cpp:176)

bool Connection::sendnow( int _cmd, const QByteArray &data )
{
    if (f_out == 0) {
	kdDebug(7017) << "write: not yet inited." << endl;
	return false;
    }

    kdDebug(7017) << "sendnow " << _cmd << endl;

    static char buffer[ 64 ];
    sprintf( buffer, "%6x_%2x_", data.size(), _cmd );

    size_t n = fwrite( buffer, 1, 10, f_out );

    if ( n != 10 ) {
	kdError(7017) << "Could not send header" << endl;
	return false;
    }

    n = fwrite( data.data(), 1, data.size(), f_out );

    if ( n != data.size() ) {
	kdError(7017) << "Could not write data" << endl;
	return false;
    }

    fflush( f_out );

    return true;
}

kdelibs'Connection::read() (./kdelibs/kio/connection.cpp:207)

int Connection::read( int* _cmd, QByteArray &data )
{
    kdDebug(7017) << "read\n";

    if (!fd_in) {
	kdDebug(7017) << "read: not yet inited" << endl;
	return -1;
    }

    static char buffer[ 10 ];

 again1:
    ssize_t n = ::read( fd_in, buffer, 10);
    if ( n == -1 && errno == EINTR )
	goto again1;

    if ( n == -1) {
	kdError(7017) << "Header read failed, errno=" << errno << endl;
    }

    if ( n != 10 ) {
      if ( n ) // 0 indicates end of file
        kdError(7017) << "Header has invalid size (" << n << ")" << endl;
      return -1;
    }

    buffer[ 6 ] = 0;
    buffer[ 9 ] = 0;

    char *p = buffer;
    while( *p == ' ' ) p++;
    long int len = strtol( p, 0L, 16 );

    p = buffer + 7;
    while( *p == ' ' ) p++;
    long int cmd = strtol( p, 0L, 16 );
    kdDebug(7017) << "read cmd " << cmd << endl;

    data.resize( len );

    if ( len > 0L ) {
	int bytesToGo = len;
	int bytesRead = 0;
	do {
	    n = ::read(fd_in, data.data()+bytesRead, bytesToGo);
	    if (n == -1) {
		if (errno == EINTR)
		    continue;
		
		kdError(7017) << "Data read failed, errno=" << errno << endl;
		return -1;
	    }
	    if (n != bytesToGo) {
		kdDebug(7017) << "Not enough data read (" << n << " instead of " << bytesToGo << ") cmd=" << cmd << "d" << endl;
	    }
	
	    bytesRead += n;
	    bytesToGo -= n;
	}
	while(bytesToGo);
    }

    *_cmd = cmd;
    kdDebug(7017) << "finished reading cmd " << cmd << endl;

    return len;
}

kdelibs'Connection::sigpipe_handler() (./kdelibs/kio/connection.cpp:275)

void Connection::sigpipe_handler(int)
{
    kdDebug() << "*** SIGPIPE ***" << endl;
    // Do nothing.
    // dispatch will return false and that will trigger ERR_SLAVE_DIED in slave.cpp
    // (or connection with app lost if this connection is in a slave)
}