/*
 * Copyright 2022-2023, ttcoder
 * All rights reserved. Distributed under the terms of the MIT license.
 */


// genode/base
#include <base/sleep.h>  // sleep_forever()
#include <base/heap.h>  // Sliced_heap
#include <root/component.h>  // Genode::Root_component

// genode/libc
#include <libc/component.h>  // Libc::with_libc()

// libc.so
#include <dirent.h>  // opendir()
#include <fcntl.h>  // open() etc
#include <poll.h>  // more modern than select()..
#include <stdio.h>
//#include <pthread.h>
//#include <sys/select.h>

// libstdc++.so
#include <map>
#include <string>
#include <vector>

// project-common
#include "BufIO.h"  // member

// libbe.so
#if 0
#include <app/Message.h>
#include <net/NetBuffer.h>
#else
typedef int port_id;
#endif

// project
#include <broker_session/broker_session.h>

namespace bro_conn {
	struct Session_component;
	struct Root_component;
	struct Main;
}

using namespace bro_conn;//


/**************/
static int debug = 1;

/**************/
enum { Invalid_Fd = -1 };


//#pragma mark - Hub -


class Hub
{
	// There is one 'Hub' singleton accessed concurrently by multiple
	// Session_component instances, but that's MT-safe since
	// all SC instances run in the same ep() thread...

public:

		// xTors
		Hub();
		
		// New, in support of forwards from Session (custom tailored code)
		void AllocateTeamAndPipe(
			Session::PseudoTeam & team_into,
			Session::Name & name_into,
			int & fd_into
			);
		void Deallocate( const Session::PseudoTeam team );
		Session::PseudoTeam
			PTeamForPort( int port ) const;
		bool IsValidPort( int port ) const;
		int SinkForTeam( Session::PseudoTeam pteam ) const;
		
		// "New" / fowards-from-interface:
		// (Session just fowards these to us, hence the identical prototypes)
		void  port_created( int port, Session::Name name );
		int   find_port( Session::Name by_name );
		void  port_deleted( int port );


private:  // Code

		void cachePipesList();


private:  // Data

		// State, 'factory'
		Session::PseudoTeam
			noncePT;
		
		// State, 'registry' of ports (across all teams)
		std::map< port_id, Session::Name >
			portNames;  // e.g. "port 9001 is named 'system:registrar'"
		
//		std::map< Session::PseudoTeam, Session::Path >
//			teamPipes;  // e.g. "pseudoteam 1 communication occurs on pipe at path /dev/pipes/app_h2_1"
			// (there is only one pipe/path per 'TEAM' used for all its ports, no need to have one pipe for each port of each team of course)

		// Config/State, list of configured named-pipes and their opened FD (if any)
		std::map< std::string, int > pipesFDs;
			// Will contain e.g.:
			// [ "/dev/bro2app/1/pipe" => -1 ]  // meaning this one is free to open()
			// [ "/dev/bro2app/2/pipe" =>  3 ]  // meaning this one is busy, must first be close()d
		
		// State, 'registry' of file-descriptors for any given team
		std::map< Session::PseudoTeam, int >
			teamFDs
			;
};



Hub::Hub()
:	noncePT( 0 )  // so that the first "team" will be 0+1 == 1 (it must NOT be 0, as 0-prefixed ports are local ports, not broker-arbitrated global IPC ports!)
	,portNames()
	,pipesFDs()
	,teamFDs()
{
	if( debug >= 5 )
		Genode::log( "Hub ctor" );
	
	// T0 for <pipesFDs>
	Libc::with_libc([&] () {
		cachePipesList();
	} );
}


void Hub::cachePipesList()
{
	// Obtain the list of existing pipes from VFS with readdir_r(), so that I can raise the
	// ceiling (increase the count) in just one step, in the .run file, with no need to keep C++ code in sync.
	//
	DIR * dir = opendir( "/dev/bro2app" );
	if( dir )
	{
		char entry_buf[1024] = {};
		dirent * entry = NULL;
		while( 0 == readdir_r(dir, (dirent*)entry_buf, &entry) && entry )
		{
			// compute e.g. "/dev/bro2app/1/pipe":
			char path[64] = {};
			{
				strcpy( path, "/dev/bro2app/" );
				strcat( path, entry->d_name );
				strcat( path, "/pipe" );
			}
			
			// T0 : insert the path (with an 'unset' FD for now)
			pipesFDs[ path ] = Invalid_Fd;  // e.g.: insert( "/dev/bro2app/1/pipe" );
			
			if( debug >= 2 )
				printf("  found tx pipe <%s>\n", path);
		}
		
		closedir( dir );
	}
	else
		Genode::error( "Cannot opendir(/dev/bro2app), IPC ports/messages won't work. Does the .run file include pipes for Broker ?" );
}


void Hub::AllocateTeamAndPipe(
			Session::PseudoTeam & team_into,
			Session::Name & name_into, /// -> "pipe_path_into"
			int & fd_into
			)
{
	// T-, so that the first "pseudo-team" will be 0+1 == 1 (it must NOT be 0)
	noncePT += 1;
	
	// T0
	team_into = noncePT;
	
	if( debug >= 2 )
		Genode::log( "handshake -> pseudo-team: ", noncePT );
	
	// We must be able to recycle previously-used pipes
	//
	int i = 0;
	auto it = pipesFDs.begin();
	for(  ; it != pipesFDs.end(); it++, i++ )
	{
		if( debug >= 2 )
			Genode::log( "  examining team <", noncePT, ">'s pipe: trying pipe <", it->first.c_str(), "> with FD: ", it->second );
		
		if( it->second < 0 ) /// <= Invalid_Fd
			//
			break;  // found a free slot !
			//
	}
	
	if( pipesFDs.end() == it )
	{
		Genode::error( "All ", pipesFDs.size(), " IPC Pipes busy, limit reached, cannot allocate new BMessage/port pipe !" );
		//
		return;
		//
	}
	;
	// T0
	Libc::with_libc([&] () {
		fd_into = open( it->first.c_str(), O_WRONLY );
	} );
	if( fd_into < 0 )
	{
		Genode::error( "cannot open() pipe <", name_into, "> in support of write_port commo -- won't be able to send BMessages to this client" );
		//
		return;
		//
	}
	// Else:
	
	// T0: out (set name only, fd_into was already set above)
	name_into = it->first.c_str();
	// T0: this.state
	teamFDs[ team_into ] = fd_into;
	it->second = fd_into;
}
#if 0
bro_conn::Session::Path
	/*bro_conn::Session_component*/Hub::pipe_for_team( Session::PseudoTeam team )
{
	auto it = teamPipes.find( team );
	
	if( it == teamPipes.end() )
	{
		Genode::warning( " pipe_for_team( ", team, " ): no pipe allocated for that team" );
		//
		return "";
	}
	
	return it->second;
}
#endif


void Hub::Deallocate( const Session::PseudoTeam team )
{
	if( debug )
		Genode::log( "Deallocate session for pseudo-team: ", team );
	
	// <portNames>:
	// unregister *all ports* for that team
	//
	for( auto it = portNames.begin(); it != portNames.end(); it++ )
	{
		if( PTeamForPort(it->first) == team )
///ToDo: Is it legal to repeatedly erase() while iterating ?
			portNames.erase( it );
	}
	
	// <teamFDs> & <pipesFDs>:
	// close and unregister the *one pipe* for that team
	//
	for( auto it = teamFDs.begin(); it != teamFDs.end(); it++ )
	{
		if( it->first == team )
		{
			if( debug )
				Genode::log( "Deallocate: close(", it->second, ")" );
			
			// if FD is valid (open'ed) for the <teamFDs> pipe, look up the pipe in <pipesFDs> as well:
			if( it->second > Invalid_Fd )
			{
				for( auto pit = pipesFDs.begin(); pit != pipesFDs.end(); pit++ )
				{
					if( pit->second == it->second )
					{
						// found the pipe -- mark it as no-longer-used:
						pit->second = Invalid_Fd;
						//
						break;  // there's only one FD per pipe
						//
					}
				}
			}
			
			Libc::with_libc([&] () {
				// T-
				if( it->second >= 0 )
					close( it->second );
			} );
			// T+
			teamFDs.erase( it );
			
			break;  // there's only one pipe per team
		}
	}
}


bro_conn::Session::PseudoTeam
	Hub::PTeamForPort( int port ) const 
{
///later: "1000 ports ought to be enough for everybody"... So we have 1000 ports per team before things start breaking... Is there a better way ? Or at least, raise the ceiling to 10k ports per team ?
	// RETURNS:
	// e.g. port 3005 -> team 3
	
	Session::PseudoTeam pt = port / 1000;
	if( pt < 1 )
	{
		// global ports start at 1000, not at 0
		Genode::error( "port is invalid or local (instead of global): ", port );
		//
		return -1;
	}
	
	return pt;
}

bool Hub::IsValidPort( int port ) const
{
	return
		portNames.find( port )
		!=
		portNames.end()
		;
}

int Hub::SinkForTeam( Session::PseudoTeam pteam ) const
{
	auto it = teamFDs.find( pteam );
	if( it == teamFDs.end() )
	{
		//Genode::error....
		//
		return -1;
	}
	
	return it->second;
}


//#pragma mark -

void Hub::port_created( int port, Session::Name name )
{
	if( debug >= 2 )
		Genode::log( "port_created <", port, "> <", name, ">" );
	
	//registerPort( name, port );
	auto it = portNames.find( port );
	if( it != portNames.end() )
		Genode::error( "registerPort: port ", port, " already exists ?! (why was this not caught upstream ?)" );
		///later: ponder this: should we return ? or proceed to below and clobber ?
	
	// T0
	portNames[ port ] = name;
}

int Hub::find_port( Session::Name by_name )
{
	//const port_id port = lookup_port( by_name );
	// if there are several ports with the same name, we return the 'first' one we encounter.
	for( auto it = portNames.begin(); it != portNames.end(); it++ )
	{
		if( it->second == by_name )
		{
			if( debug >= 2 )
				Genode::log( "find_port <", by_name, "> -> ", it->first );
			
			return it->first;
		}
	}
	
	if( debug >= 2 )
		Genode::log( "find_port <", by_name, "> -> Not Found !" );
	
	return -1;
}

void Hub::port_deleted( int port )
{
	auto it = portNames.find( port );
	
	if( it == portNames.end() )
	{
		Genode::error( "port_deleted: port ", port, " not found... already deleted..? That case would have been caught upstream in libroot though..." );
		//
		return;
	}
	
	// T0
	portNames.erase( it );
}



//#pragma mark - Session_component -


struct/**/ bro_conn::Session_component : Genode::Rpc_object<Session>
{
	/*
		A session/connection to a haiku-on-genode application, for us to
		respond to said app's RPC requests re. *inbound* port messages (app-to-broker),
		and with a (connection-unique) unix pipe through which to send
		*outbound* (broker-to-recipient-app) port data.
		FWIW, the data that transits through ports is typically flattened BMessage's.
		So a Haiku app can have BMessages sent to (and received from) other apps,
		through this server.
	*/

public:  // Code

		// xTors
		
		Session_component( Hub & hub, Genode::Allocator & alloc );
		~Session_component();
		
		
		// 'Session' overrides
		
		PseudoTeam  id_for_team() override;
		Path  pipe_for_team() override;
		void  port_created( int port, Name name ) override;
		int   find_port( Name by_name ) override;
		int   writeport_header( long discriminant, int port, int code, int overall_size ) override;
		int   writeport_chunk(  long discriminant, int port, Session::BufData buf, int chunk_size ) override;
		void  port_deleted( int port ) override;
		int pipeFiledescFor( int port );///

		
private:  // Data

		// Context
		Hub & hub;  // data *shared* between all session/connections.
		
		// State
		std::map< long, BufIO* > bufByThread;
			// writeport is not atomic, so several concurrent write_port multi-RPC-call transactions might end up "multiplexed",
			// we use the (long int) thread/discriminant to dispatch each chunk to its own transaction (its own accumulator/buffer).
		
		// Ctor-retrieved config, re our client counterpart
		PseudoTeam clientTeam;
		// to provide client with name of pipe to (read) at its end:
		Name pipeName;
		// for us to (write) at our end of the pipe:
		int pipeFd;  // we write() to this FD to Tx write_port data on its second-leg voyage.
		
		// System
		Genode::Allocator & memAlloc;  // to allocate BufIO's
};




bro_conn::Session_component::Session_component( Hub & _hub, Genode::Allocator & alloc )
:	hub( _hub )
	,bufByThread()
	,clientTeam( -1 )
	,pipeName()
	,pipeFd( -1 )
	,memAlloc( alloc )
{
	// open() the pipe and register it
	hub.AllocateTeamAndPipe(
		clientTeam,
		pipeName,
		pipeFd
		);
	
	if( debug >= 2 )
		Genode::log( "* New SessionComponent team <", clientTeam, "> pipe-path <", pipeName, "> FD <", pipeFd, ">" );
}


bro_conn::Session_component::~Session_component()
{
	// close() the pipe and un-register it, then un-register all ports owned by that team
	hub.Deallocate( clientTeam );
}


bro_conn::Session::PseudoTeam
	bro_conn::Session_component::id_for_team()
{
	return clientTeam;
}

bro_conn::Session::Path
	bro_conn::Session_component::pipe_for_team()
{
	return pipeName;
}

void bro_conn::Session_component::port_created( int port, Session::Name name )
{
	hub.port_created( port, name );
}

int bro_conn::Session_component::find_port( Session::Name by_name )
{
	return hub.find_port( by_name );
}

void bro_conn::Session_component::port_deleted( int port )
{
	hub.port_deleted( port );
}


int bro_conn::Session_component::writeport_header( long discriminant, int port, int code, int complete_size )
{
	if( debug >= 2 )
	{
		Genode::log( "writeport_header <", port, "> <", code, "> ", complete_size, " overall data bytes expected" );
//		Libc::with_libc([&] () {
//			printf( " (discriminant: 0x%lx)\n", discriminant );
//		});
	}
	
	int tx = pipeFiledescFor( port );
	if( tx < 0 )
		//
		return tx;
		//
	
#if 0
	bool complete = false;
	;
	Libc::with_libc([&] ()
	{
		// T0 a..b..c..
		int sent = write( tx, &port, sizeof(port) );
		if( sizeof(port) == sent )
		{
			sent = write( tx, &code, sizeof(code) );
			if( sizeof(code) == sent )
			{
				sent = write( tx, &complete_size, sizeof(complete_size) );
				
				// T+ / success
				if( sizeof(complete_size) == sent )
					complete = true;
			}
		}
	} );
	
	if( false == complete )
		Genode::error( "incomplete write()s to (port) pipe (failure mid-stream ?)" );
	
	return complete
		? B_OK
		: B_ERROR
		;
#else
	//auto * tagged_buf = bufByThread[ discriminant ];//tmp_buf;
	//auto * accu = tagged_buf;//tagged_buf->second
	if( bufByThread[ discriminant ] )//accu != nullptr )
	{
		Genode::error( "writeport_Header called, but the Previous msg buf was not yet sent! Discarding all" );
		delete bufByThread[ discriminant ]; bufByThread[ discriminant ] = nullptr;//tagged_buf->second = nullptr;
		return -1;
	}
	//tagged_buf->first = discriminant;
	//tagged_buf->second = new BufIO( memAlloc );
	auto * accu = new BufIO( memAlloc );
	bufByThread[ discriminant ] = accu;
	
	//Genode::log(" t- size: ", tagged_buf->second->BufferLength() );
	accu->Append( &port, sizeof(port) );
	accu->Append( &code, sizeof(code) );
	accu->Append( &complete_size, sizeof(complete_size) );
	//Genode::log(" t+ size: ", tagged_buf->second->BufferLength() );
	
///ToDo: bleh ! refactor into a Flush() method, called from here and from the next method
// or maybe client-side code in Remote::WritePort() could call writeport_chunk(null, 0) to save us the trouble ?
	// hack: we need to call write() in the special case of zero-size (since no one else will call it, in that specific case)...
	if( complete_size <= 0 )
		writeport_chunk( discriminant, port, Session::BufData(), 0 );
	
	if( debug >= 2 )
		Genode::log( "writeport_header -> B_OK" );
	
	return B_OK;
#endif
}


int bro_conn::Session_component::writeport_chunk( long discriminant, int port, Session::BufData buf, int chunk_size )
{
	if( debug >= 3 )
	{
		Genode::log( "writeport_chunk <", port, "> ", chunk_size, " bytes in this one chunk" );
//		Libc::with_libc([&] () {
//			printf( " (discriminant: 0x%lx)\n", discriminant );
//		});
	}
	
	int tx = pipeFiledescFor( port );
	if( tx < 0 )
		//
		return tx;
		//
	
#if 0
	int sent = -1;
	;
	Libc::with_libc([&] () {
		sent = write( tx, buf.buf, chunk_size );
	} );
	
	if( debug >= 3 )
		Genode::log( "  ..success : one chunk sent" );
	
	return sent;

#else

	auto * accu = bufByThread[ discriminant ];//tmp_buf;
	
	if( accu == nullptr )
	{
		Genode::error( "writeport_Chunk without prior writeport_Header ! Ignoring..." );
		
		return -1;
	}
	
	//Genode::log(" T- size: ", tagged_buf->second->BufferLength() );
	accu->Append( buf.buf, chunk_size );
	//Genode::log(" T+ size: ", tagged_buf->second->BufferLength() );
	
	// Calculate "flush" threshold:
	size_t
		threshold = 0;
	if( accu->BufferLength() < 3*sizeof(int) )
	{
		Genode::error( "writeport_Chunk without (proper) prior writeport_Header ! Clearing state..." );
		
		// clear:
		delete bufByThread[ discriminant ];
		bufByThread[ discriminant ] = nullptr;
		
		return -1;
	}
	else
	{
		// Calculate expected buffer size (payload + overhead, i.e. all-chunks + 3 integers)
		const int * arr = (const int*)accu->Buffer();
		threshold = arr[ 2 ] + 3*sizeof(int); ///clean this up: use a struct/union
	}
	
	//Genode::log( "  comparing ", tagged_buf->second->BufferLength(), " versus ", threshold );
	
	//
	// Is this the *last* chunk, or should we wait before flushing ? :
	//
	
	if( accu->BufferLength() < threshold )
	{
		/* Torture Test/Edge-case test, to detect the race condition:
		// silly replacement for snooze()
		static volatile int dummy = 0;
		for(int i = 0; i < 100999000; i++) { if(dummy!=0)break; }
		*/
		
		// we're not done yet, wait for more chunk(s) before flushing
		return chunk_size;
		//
	}
	else if( accu->BufferLength() == threshold )
	{
		// Flush !
		
		int sent = -1;
		
		Libc::with_libc([&] () {
			sent = write( tx, accu->Buffer(), accu->BufferLength() );
		} );
		
		// T-
		const int res = (unsigned(sent) != accu->BufferLength())
			? sent  // error case
			: chunk_size  // success case, and caller is interested in current chunk's sent status, not in overall sent status
			;
		
		// T0
		delete bufByThread[ discriminant ];//tagged_buf->second;
		bufByThread[ discriminant ] = nullptr;
		
		if( debug >= 2 )
			Genode::log( "writeport_chunk -> wrote ", res, " bytes to FD ", tx );
		
		// T+
		return res;
	}
	else  // BufferLength() > expected threshold value ! :
	{
		Genode::error( "writeport buffer overshot, did we mix up chunks from incompatible streams ? Clearing state..." );
		
		// clear:
		delete bufByThread[ discriminant ];
		bufByThread[ discriminant ] = nullptr;
		
		return -1;
	}
	
	return -1; //NotReached
#endif
}


///ToDo: move this to class Hub ?
int bro_conn::Session_component::pipeFiledescFor( int port )
{
	// Sanity check via <portNames> (otherwise we would never return B_BAD_PORT_ID, based on <teamPipes> alone):
	if( false == hub.IsValidPort(port) )
		//
		return B_BAD_PORT_ID;
		//
	
	// First we calculate a "team" number to look up:
	auto pteam = hub.PTeamForPort( port );
	
	if( pteam < 1 )  // global ports start at 1000, i.e. at pseudo-team 1
	{
		Genode::error( "write_port() to port ", port, ": does not correspond to a valid, and local (instead of global) pseudo-team, can't find which pipe to use to write to that port" );
		//
		return B_BAD_PORT_ID;
	}
	
	return
		hub.SinkForTeam( pteam );
}

#if 0
	_d_ this is limited to tiny BMessages that fit in one RPC call:
int bro_conn::Session_component::write_port( int global_port, int code, Session::BufData buf, int size )
{
	if( debug )
		Genode::log( "write_port <", global_port, "> <", code, "> ", size, " bytes" );
	//Genode::log( "write-port to port ", port, " with code ", code, " size ", size );
	
	// Sanity check via <portNames> (otherwise we would never return B_BAD_PORT_ID, based on <teamPipes> alone):
	if( false == hub.IsValidPort(global_port) )//portNames.find(global_port) == hub.portNames.end() )
		//
		return B_BAD_PORT_ID;
		//
	
	// the recipient is probably *another* Session_component, not <this> one
	// (otherwise pseudo-kernel would have taken a shortcut and delivered the
	// payload directly, instead of indirecting through Broker here) ; so
	// determine which session to use (or rather, what the File Descriptor of recipient
	// session's pipe is).
	
	int tx = -1;
	{
		// First we calculate a "team" number to look up:
		auto pteam = hub.PTeamForPort( global_port );
		
		if( pteam < 1 )  // global ports start at 1000, i.e. at pseudo-team 1
		{
			Genode::error( "write_port() to port ", global_port, ": does not correspond to a valid, and local (instead of global) pseudo-team, can't find which pipe to use to write to that port" );
			//
			return B_BAD_PORT_ID;//-1;
		}
		
		tx = hub.SinkForTeam( pteam );
	}
	if( tx < 0 )
	{
		Genode::error( "write_port to port ", global_port, ": pipe was not allocated in ctor, maybe we had run out of pipes ?" );
		//
		return -2;
	}
	
	int recipient_local_port = global_port;//-1;
	{
/*
_d_ !! we want to preserve the port number!
		if( global_port <= 999 )
			Genode::error("this is a local port number, not a global port number !");
		else if( global_port <= 1999 )
			recipient_local_port = global_port - 1000;
		else if( global_port <= 2999 )
			recipient_local_port = global_port - 2000;
		else
			Genode::error("port_level NOT IMPLEMENTED YET (easy !)");
*/
	}
	/*
	auto it = teamPipes.find( pteam );
	if( it == teamPipes.end() )
	{
		Genode::error( "write_port to port ", global_port, ": pipe was not allocated in team_started(), maybe we had run out of pipes ?" );
		//
		return -2;
	}
	*/
	bool completion = false;
	
	//int sent = 0;
	Libc::with_libc([&] ()
	{
		// T0 a..b..c..
		int sent = write( tx, &recipient_local_port, sizeof(recipient_local_port) );
		if( sizeof(recipient_local_port) == sent )
		{
			sent = write( tx, &code, sizeof(code) );
			if( sizeof(code) == sent )
			{
				sent = write( tx, &size, sizeof(size) );
				if( sizeof(size) == sent )
				{
					sent = write( tx, buf.buf, size );
					
					// T+ / success
					if( size == sent )
					{
						if( debug >= 3 )
							Genode::log( "  ..success : every single write() call completed in full" );
						
						//
						completion = true;//return B_OK;
						//
					}
				}
			}
		}
	});
	
	if( false == completion )
		Genode::error( "incomplete write()s to (port) pipe (failure mid-stream ?)" );
	
	return completion
		? B_OK
		: B_ERROR
		;//sent;  // success
}
#endif



//#pragma mark - Root_component -


class bro_conn::Root_component : public Genode::Root_component<Session_component>
{
public:
		Root_component(
			Genode::Env & env_arg,
			Genode::Entrypoint & ep,
			Genode::Allocator & alloc
			)
		:	Genode::Root_component<Session_component>( ep, alloc )
			,env( env_arg )
			,unifiedHub()
		{
			//Genode::log("creating Broker root component");
		}

protected:  // Code

		// Override in Genode::Root_component... (or is it in Session_component?)
		Session_component *_create_session( const char * str ) override
		{
			// this will display e.g.:
			//		<< creating Broker session -- str: diag=0, label="_test_client -> ", ram_quota=1229, cap_quota=3 >>
			if( debug )
			{
				Genode::log( "creating Broker session -- str: ", str );
				//Genode::log( "  used caps/mem:  ", env.pd().used_caps().value, ", ", env.pd().used_ram().value, " bytes" );
			}
			
			return new (md_alloc()) Session_component( unifiedHub, *md_alloc() );
		}

private:  // Data

		Genode::Env &  // we don't really need to keep an env ref around, except for debug tracing of used_caps() et al
			env;
		
		Hub unifiedHub;
};



//#pragma mark - Broker -


class Broker
{
public:
	Broker( Genode::Env & env_arg );
	void Run();
	
#if 0
private:  // Code
	void digestInbound();
	void registerPort( const char * name, port_id id );
	port_id lookup_port( const char * name );
#endif


private:  // Data

#if 0
	// I/O: buf
	BNetBuffer ser; //or "inboundTraffic"..
	// State
	std::map< port_id, std::string > nameForPort; // or "portNames"
	// I/O: outbound FD
	int tx;
#else
	Genode::Sliced_heap
		sliced_heap;
	bro_conn::Root_component
		root;
#endif
};



Broker::Broker( Genode::Env & env )
:	sliced_heap( env.ram(), env.rm() )
	,root( env, env.ep(), sliced_heap )
{
	/*
	 * Create a RPC object capability for the root interface and
	 * announce the service to our parent.
	 */
	env.parent().announce(
		env.ep().manage( root )
		);
	
	if( debug >= 5 )
		Genode::log( "ctor and announce() done" );
}


#if 0
	_d_ old code (using inbound pipes instead of RPC calls):

#define CHECK(x) if(x) { /*puts( "(pass) " #x );*/ } else { puts( " ***** FAIL! : " #x ); }

void Broker::registerPort( const char * name, port_id id )
{
	// don't check for _name_ dupes, those are allowed (unlike port_ids which are unique)
	
	auto it = nameForPort.find( id );
	if( it != nameForPort.end() )
		Genode::error( "registerPort: port ", id, " already exists ?!" );
		// xx if ID exists, error or just clobber ?
	
	// T0
	nameForPort[ id ] = name;
}

port_id
	Broker::lookup_port( const char * name )
{
	for( auto it = nameForPort.begin(); it != nameForPort.end(); it++ )
	{
		if( it->second == name )
			return it->first;
	}
	
	return -1;
}

void Broker::digestInbound()
{
	int32 code = 0;
	{
		CHECK( B_OK == ser.RemoveInt32(code) );
		//Genode::warning( "GOT Code: ", code );
	}
	
	switch( code )
	{
	//xx Handshake : allocate private pipe
		case bro::HANDSHAKE:
		{
	static int id = 1;
			//if( debug )
			Genode::log( " .. handshake .." );
			
			tx = open( "/dev/downstream", O_WRONLY );
			CHECK( tx >= 0 );
			
			const int value = id++;
			CHECK( sizeof(value) == write(tx, &value, sizeof(value)) );
	//move this one to "deleted_port()" :		CHECK( close(tx) == 0 );
		}
		break;
		
		case bro::CREATED_PORT:
		{
			port_id id = 0;
			char name[ B_OS_NAME_LENGTH ] = {};
			
			CHECK( B_OK == ser.RemoveInt32(id) );
			CHECK( B_OK == ser.RemoveString(name, sizeof(name)) );
			
			//if( debug )
			Genode::log( " .. register(", (const char*)name, ", ", id, ")" );
			registerPort( name, id );
		}
		break;
		
		case bro::FIND_PORT:
		{
			char name[ B_OS_NAME_LENGTH ] = {};
			CHECK( B_OK == ser.RemoveString(name, sizeof(name)) );
			//Genode::warning( "GOT Name: ", (const char*)name );
			
			const port_id port = lookup_port( name );
			//if( debug )
			Genode::log( " .. lookup(", (const char*)name, ") => ", port );
			
	//						int tx = open( "/dev/downstream", O_WRONLY );
	//		CHECK( tx >= 0 );
			CHECK( sizeof(port) == write(tx, &port, sizeof(port)) );
	//						CHECK( close(tx) == 0 );
		}
		break;
		
		case bro::WRITE_PORT:
		{
			// Tx to "client"
			//
Genode::error("n/i");
//xx  // client ? which one ? (should maintain a map<port : pipe>...) -- including for PORT_FOR_REGISTRAR (hardcoded)..
//xx NOT DOWNSTREAM of requester-team! downstream/H2 of actual target-team !
//						int tx = open( "/dev/downstream", O_WRONLY );// xx !
//			CHECK( tx >= 0 );
//no, not <tx> !	CHECK( 3 == write(tx, "abc", 3) );
//						CHECK( close(tx) == 0 );
		}
		break;
		
		default:
			Genode::error( "unknown opcode: ", code );
	}
	
	// no need to flush, since we Remove() stuff..
	//ser = BNetBuffer();
}
#endif


void Broker::Run()
{
#if 0
	// Rx from "client"
	//
//xx rename to "temp_dispatch_request" and "temp_dispatch_response"
//xx jam t8: add a call to "BMessenger(app/x-vnd-test).SendMessage(0x123)" to *trace* invocations of write_port() !
//implement 'transport' of source write_port() to destination (to then be picked up by read_port())
	int rx = open( "/dev/upstream", O_RDONLY );
	CHECK( rx >= 0 );
	
	for( ; ; ) //+ while (num_open_fds > 0)..
	{
		if( debug )
			Genode::log( "---------- ready --------" );
		
#if 0
		// select():
		
		fd_set readfds;
		{
			FD_ZERO( &readfds );
			FD_SET( rx, &readfds );
		}
		int yield = select( rx+1, &readfds, NULL, NULL, NULL );
		if( yield < 0 )
			perror("select()");
		
		// if( FD_ISSET(rx, &readfs) ) ..
#endif
		struct pollfd pfds[] = {
			{ .fd = rx, .events = POLLIN, .revents = 0 }
		};
		int nfds = 1; //+ B_COUNT_OF( pfds )..
		
		// poll()
		//
#if 1
		int ready = poll( pfds, nfds, -1 );
		if( ready < 0 )
			perror( "poll()" );
#else
		int ready = 1;
#endif
		
		if( debug )
			printf( "  --> poll(): Ready: fd %d\n", ready );
		
		// Find out *which* fd got fed
		//
		for( int i = 0; i < nfds; i++)
		{
			if( pfds[i].revents == 0 )
				continue;  // not this one...
			
			// This one! Let's read:
			//xx check status for POLLIN/POLLHUP/POLLERR...
//we might have TWO or more msgs in the pipe, don't read-then-throw-away !
#if 0
			char buf[256] = {};//xxxx
			const int got = read(
				pfds[i].fd,
				buf,
				sizeof( buf )
				);
			
			if( got == 0 )
				break;
			if( got < 0 )
			{
				Genode::error( "rx error: ", got );
				break;
			}
#else
			int got = -1;
//			char buf[2 /**/] = {};
//			while( (got = read(pfds[i].fd, buf, sizeof(buf))) > 0 )
//xxxxxx -- The above blocks on read().... do the below for now, but cannot be a long term solution (big messages will result in blocking again !)
			char buf[256] = {};
			if( (got = read(pfds[i].fd, buf, sizeof(buf))) > 0 )
			{
				ser.AppendData( buf, got );
			}
			
			while( ser.Size() > 0 )
				digestInbound();
#endif
			
			//Genode::log( "Broker Rx:" );
			{
				/*
				BMessage inbound;
				const status_t built =
					inbound.Unflatten( buf );
				
				if( B_OK == built )
					inbound.PrintToStream();
				else
					Genode::error( "  cannot unflatten this buffer: ", (const char*) buf );
				*/
				
				/*
				BNetBuffer ser;
				{
					ser.AppendData( buf, got );
				}
				*/
			}
			
		}
	}
	Genode::error( "NOT SUPPOSED TO QUIT -- this will destroy 'root' RPC Handler !!" );
//?	CHECK( close(rx) == 0 );
#endif
	
	Genode::sleep_forever();
}



//#pragma mark -

#if 0
static
void * thread_pipe_server( void* )
{
extern Genode::Env * HogEnv();///
	Broker broker( *HogEnv() );
	
	broker.Run();
	
//?	pthread_exit(NULL);
	return NULL;
}



int main()
{
	// --- "server" ---
	
//	Libc::with_libc([] () {
//		pthread_t server = 0;
//		CHECK( 0 ==  pthread_create(&server, nullptr, thread_pipe_server, nullptr) );
//	});
	thread_pipe_server( nullptr );
	
	return 0;
}

#else

void Libc::Component::construct( Libc::Env & env )
{
	if( debug >= 5 )
		Genode::log( "Libc::Component::construct" );
	
	// !
	static	Broker broker( env );
}

#endif

