client.cc

This Source-Navigator, an IDE for C/C++/Fortran/Java/Tcl/PHP/Python and a host of other languages.

		if ( delim )
			*delim = '\0';

		// Find separation between event name and value.
		delim = strchr( buf, ' ' );
		Value* result;

		if ( ! delim )
			result = error_value();

		else
			{
			*delim++ = '\0';
			result = new Value( delim );
			}

		last_event = new GlishEvent( strdup( buf ), result );
		}

	if ( last_event )
		{
		const char* name = last_event->name;

		if ( streq( name, "terminate" ) )
			{
			Unref(last_event);
			last_event = 0;
			}

		else if ( name[0] == '*' )
			{
			Value* v = last_event->value;

			if ( streq( name, "*sync*" ) )
				Reply( false_value );

			else if ( streq( name, "*link-sink*" ) )
				BuildLink( v );

			else if ( streq( name, "*rendezvous-orig*" ) )
				RendezvousAsOriginator( v );

			else if ( streq( name, "*rendezvous-resp*" ) )
				RendezvousAsResponder( v );

			else if ( streq( name, "*unlink-sink*" ) )
				UnlinkSink( v );
			}
		}

	return last_event;
	}

void Client::FD_Change( int /* fd */, int /* add_flag */ )
	{
	}

void Client::HandlePing()
	{
	}

void Client::UnlinkSink( Value* v )
	{
	int is_new;
	EventLink* el = AddOutputLink( v, 1, is_new );

	if ( ! el )
		return;

	if ( is_new )
		PostEvent( "error", "no link to %s", el->ID() );

	el->SetActive( 0 );
	}

void Client::BuildLink( Value* v )
	{
	int is_new;
	EventLink* el = AddOutputLink( v, 0, is_new );

	if ( ! el )
		return;

	if ( ! is_new )
		{ // No need to forward, we've already built the link.
		el->SetActive( 1 );
		return;
		}

	char* sink_id;
	if ( ! v->FieldVal( "sink_id", sink_id ) )
		fatal->Report( "bad internal link event" );

	int fd = remote_sinks[sink_id];
	delete sink_id;

	// Minor-league bug here.  fd=0 is legal, though it means that
	// somehow stdin got closed.  But even in that case, we'll just
	// miss an opportunity to reuse the connection, and instead create
	// a new, separate connection.
	if ( fd )
		{ // Already have a connection, piggyback on it.
		el->SetFD( fd );
		el->SetActive( 1 );
		return;
		}

	// We have to create a new AcceptSocket each time, otherwise
	// there's a race in which we send out rendezvous events to
	// B and then to C, but C gets the event first and connects
	// to the accept socket in lieu of B.
	AcceptSocket* accept_socket;

	int is_local;
	if ( ! v->FieldVal( "is_local", is_local ) )
		fatal->Report( "internal link event lacks is_local field" );

	if ( is_local )
		{
		accept_socket = new AcceptSocket( 1 );

		char* path = local_socket_name( accept_socket->FD() );
		el->SetPath( path );
		v->SetField( "path", path );
		}

	else
		{ // Use an ordinary socket.
		accept_socket = new AcceptSocket();

		// Include information in the event about how to reach us.
		v->SetField( "host", local_host );
		v->SetField( "port", accept_socket->Port() );
		}

	// So we can recover this socket when the interpreter reflects
	// this rendezvous event reflected back to us.
	v->SetField( "accept_fd", accept_socket->FD() );

	PostEvent( "*rendezvous*", v );
	}

EventLink* Client::AddOutputLink( Value* v, int want_active, int& is_new )
	{
	char* event_name;
	char* new_name;
	char* task_id;

	if ( ! v->FieldVal( "event", event_name ) ||
	     ! v->FieldVal( "new_name", new_name ) ||
	     ! v->FieldVal( "sink_id", task_id ) )
		fatal->Report( "bad internal link event" );

	event_link_list* l = output_links[event_name];

	if ( ! l )
		{ // Event link doesn't exist yet.
		l = new event_link_list;
		output_links.Insert( event_name, l );
		}

	else
		{ // Look for whether this event link already exists.
		delete event_name;

		loop_over_list( *l, i )
			{
			EventLink* el = (*l)[i];

			if ( el->Active() == want_active &&
			     streq( el->Name(), new_name ) &&
			     streq( el->ID(), task_id ) )
				{ // This one fits the bill.
				delete new_name;
				delete task_id;

				is_new = 0;

				return el;
				}
			}
		}

	EventLink* result = new EventLink( task_id, new_name, want_active );
	l->append( result );

	is_new = 1;

	return result;
	}

void Client::RendezvousAsOriginator( Value* v )
	{
	int is_new;
	EventLink* el = AddOutputLink( v, 0, is_new );

	if ( ! el || is_new )
		fatal->Report( "bad internal link event" );

	int output_fd;

	int accept_fd;
	if ( ! v->FieldVal( "accept_fd", accept_fd ) )
		fatal->Report( "bad internal link event" );

	char* sink_id;
	if ( ! v->FieldVal( "sink_id", sink_id ) )
		fatal->Report( "bad internal link event" );

	int sink_fd = remote_sinks[sink_id];
	if ( sink_fd )
		{
		// Already have a socket connection to this remote sink.
		// Even though BuildLink endeavors to avoid sending out
		// a *rendezvous* if such a connection already exists,
		// it's possible for a second *link-sink* to come in before
		// we've finished processing the first, in which case the
		// second call to BuildLink won't see that we already have
		// a socket connection.  So we check here, too.
		output_fd = sink_fd;
		delete sink_id;

		// If this was a local socket, get rid of the corresponding
		// file.
		char* path;
		if ( v->FieldVal( "path", path ) )
			{
			if ( unlink( path ) < 0 )
				PostEvent( "error",
					"can't unlink local socket sink %s",
						path );

			delete path;
			}
		}

	else
		{ // Create a new socket connection.
		int is_local;
		if ( ! v->FieldVal( "is_local", is_local ) )
			fatal->Report( "bad internal link event" );

		output_fd = is_local ?
			accept_local_connection( accept_fd ) :
			accept_connection( accept_fd );

		if ( output_fd < 0 )
			fatal->Report( "socket accept for link failed" );

		remote_sinks.Insert( sink_id, output_fd );
		}

	close( accept_fd );

	el->SetFD( output_fd );
	el->SetActive( 1 );
	}

void Client::RendezvousAsResponder( Value* v )
	{
	char* source_id;
	if ( ! v->FieldVal( "source_id", source_id ) )
		fatal->Report( "bad internal link event" );

	if ( remote_sources[source_id] )
		{
		// We're all set, we already select on this input source.
		delete source_id;
		return;
		}

	int input_fd;

	int is_local;
	if ( ! v->FieldVal( "is_local", is_local ) )
		fatal->Report( "bad internal link event" );

	if ( is_local )
		{
		input_fd = get_local_socket();

		char* path;
		if ( ! v->FieldVal( "path", path ) )
			fatal->Report( "bad internal link event" );

		if ( ! local_connection( input_fd, path ) )
			fatal->Report( "couldn't connect to local socket",
					path );

		// Now that we've rendezvous'd, remove the Unix-domain
		// socket from the file system namespace.

		if ( unlink( path ) < 0 )
			PostEvent( "error", "can't unlink local socket sink %s",
					path );

		delete path;
		}

	else
		{
		char* host;
		int port;

		if ( ! v->FieldVal( "host", host ) ||
		     ! v->FieldVal( "port", port ) )
			fatal->Report( "bad internal link event" );

		input_fd = get_tcp_socket();

		if ( ! remote_connection( input_fd, host, port ) )
			{
			PostEvent( "error", "can't open socket sink %s", host );
			return;
			}

		delete host;
		}

	input_links.append( input_fd );
	FD_Change( input_fd, 1 );

	remote_sources.Insert( source_id, input_fd );
	}

void Client::SendEvent( const GlishEvent* e, int sds )
	{
	if ( no_glish )
		return;

	const char* name = e->name;
	const Value* value = e->value;

	if ( have_interpreter_connection )
		{
		event_link_list* l = output_links[name];

		if ( l )
			{
			int did_send = 0;

			loop_over_list( *l, i )
				{
				EventLink* el = (*l)[i];

				if ( el->Active() )
					{
					send_event( el->FD(), el->Name(),
							e, sds );
					did_send = 1;
					}
				}

			// If we didn't send any events then it means that
			// all of the links are inactive.  Forward to
			// the interpreter instead.
			if ( ! did_send )
				send_event( write_fd, e->name, e, sds );
			}
		else
			send_event( write_fd, e->name, e, sds );
		}

	else if ( value )
		message->Report( name, " ", value );

	else
		message->Report( name, " <no value>" );
	}

static int read_buffer( int fd, char* buffer, int buf_size )
	{
	int status = read( fd, buffer, buf_size );

	if ( status == 0 )
		return 0;

	else if ( status < 0 )
		{
		if ( errno != EPIPE && errno != ECONNRESET )
			error->Report( prog_name,
					": problem reading buffer, errno = ",
					errno );
		return 0;
		}

	else if ( status < buf_size )
		return read_buffer( fd, buffer + status, buf_size - status );

	return 1;
	}

static int write_buffer( int fd, char* buffer, int buf_size )
	{
	int status = write( fd, buffer, buf_size );

	if ( status < 0 )
		{
		if ( errno != EPIPE && errno != ECONNRESET )
			error->Report( prog_name,
					": problem writing buffer, errno = ",
					errno );
		return 0;
		}

	else if ( status < buf_size )
		return write_buffer( fd, buffer + status, buf_size - status );

	return 1;
	}


GlishEvent* recv_event( int fd )
	{
	struct event_header hdr;

	if ( ! read_buffer( fd, (char*) &hdr, sizeof( hdr ) ) )
		return 0;

	hdr.flags = ntohl( hdr.flags );
	int size = (int) ntohl( hdr.event_length );

	Value* result;

	if ( hdr.flags & GLISH_STRING_EVENT )
		{
		char* value = new char[size+1];
		if ( size > 0 && ! read_buffer( fd, value, size ) )
			{
			error->Report( prog_name,
		": read only partial string event value from socket, errno = ",
					errno );
			delete value;
			return 0;
			}

		else
			{
			value[size] = '\0';
			result = split( value );
			}

		delete value;
		}

	else
		{
		int sds = (int) sds_read_open_fd( fd, size );

		if ( sds <= 0 )
			{
			if ( sds != SDS_FILE_WR )
				error->Report( prog_name,
				": problem reading event, SDS error code = ",
						sds );
			return 0;
			}

		else
			result = read_value_from_SDS( sds,
					int( hdr.flags & GLISH_OPAQUE_EVENT ) );
		}

	GlishEvent* e = new GlishEvent( strdup( hdr.event_name ), result );
	e->SetFlags( int( hdr.flags ) );
	return e;
	}


static int send_event_header( int fd, int flags, int length, const char* name )
	{
	// The following is static so that we don't wind up copying
	// uninitialized memory; makes Purify happy.
	static struct event_header hdr;

	hdr.flags = htonl( (u_long) flags );
	hdr.event_length = htonl( (u_long) length );

	if ( strlen( name ) >= MAX_EVENT_NAME_SIZE )
		warn->Report( prog_name,
				": event name \"", name, "\" truncated to ",
				MAX_EVENT_NAME_SIZE - 1, " characters" );

	strncpy( hdr.event_name, name, MAX_EVENT_NAME_SIZE - 1 );
	hdr.event_name[MAX_EVENT_NAME_SIZE - 1] = '\0';

	if ( ! write_buffer( fd, (char*) &hdr, sizeof( hdr ) ) )
		return 0;

	return 1;
	}

void send_event( int fd, const char* name, const GlishEvent* e, int sds )
	{
	const Value* value = e->value;
	int event_flags = e->Flags();

	if ( value )
		value = value->Deref();

	if ( value && value->Type() == TYPE_STRING && value->Length() == 1 &&
	     ! value->AttributePtr() )
		{
		const_args_list a;
		a.append( value );
		char* string_val = paste( &a );

		int size = strlen( string_val );
		event_flags |= GLISH_STRING_EVENT;

		if ( send_event_header( fd, event_flags, size, name ) )
			(void) write_buffer( fd, string_val, size );

		delete string_val;
		}

	else
		{
		if ( value && value->Type() == TYPE_OPAQUE )
			{
			sds = value->SDS_IndexVal();
			value = 0;
			}

		del_list d;

		if ( sds >= 0 )
			event_flags |= GLISH_OPAQUE_EVENT;

		else
			{
			if ( ! value )
				value = new Value( glish_false );

			sds = (int) sds_new( (char*) "" );

			if ( sds < 0 )
				error->Report( prog_name,
				": problem posting event, SDS error code = ",
						sds );

			value->AddToSds( sds, &d );
			}

		int size = (int) sds_fullsize( sds );

		if ( send_event_header( fd, event_flags, size, name ) )
			{
			if ( sds_write2fd( fd, sds ) != sds &&
			     sds_error != SDS_FILE_WR )
				error->Report( prog_name,
			": problem sending event on socket, SDS error code = ",
						(int) sds_error );
			}

		if ( value )
			{
			sds_destroy( sds );
			sds_discard( sds );

			delete_list( &d );
			}
		}
	}