bus: use json for data transport, new plugins

control/sys.txt

@@ -49,7 +49,7 @@ loadPlugins 2
 # loadPlugins_list <list>
 #   if loadPlugins is set to 2, this comma-separated list of plugin names (filename without the extension)
 #   specifies which plugin files to load at startup or when the "plugin load all" command is used.
-loadPlugins_list macro,profiles,breakTime,raiseStat,raiseSkill,map,reconnect,eventMacro,item_weight_recorder
+loadPlugins_list macro,profiles,breakTime,raiseStat,raiseSkill,map,reconnect,eventMacro,item_weight_recorder,bus_hook,bus_party,bus_command
 
 # skipPlugins_list <list>
 #   if loadPlugins is set to 3, this comma-separated list of plugin names (filename without the extension)

plugins/bus_command/bus_command.pl

@@ -0,0 +1,59 @@
+package OpenKore::Plugins::BusCommand;
+###############################################################################
+# Plugin to allow console commands to be sent via bus.
+# Depends on the bus_hook plugin.
+
+use strict;
+
+use Globals qw( $char %config $field $net );
+use Utils qw( &existsInList );
+
+our $name = 'bus_command';
+
+Plugins::register( $name, "$name plugin", \&Unload, \&Unload );
+
+my $hooks = Plugins::addHooks(    #
+	[ 'bus/recv/RUN_COMMAND' => \&onBusRecvRunCommand ],
+);
+
+sub Unload {
+	Plugins::delHooks( $hooks );
+}
+
+sub onBusRecvRunCommand {
+	my ( undef, $args ) = @_;
+
+    my $allow = 1;
+	$allow = 0 if $args->{group} && !check_group( $args->{group} );
+	$allow = 0 if $args->{party} && !( $char && $char->{party} && $char->{party}->{name} eq $args->{party} );
+	if ( !$allow ) {
+		Log::debug( "[$name] Received and ignored command: $args->{command}\n", $name );
+		return;
+	}
+
+	Log::debug( "[$name] Received command: $args->{command}\n", $name );
+	Commands::run( $args->{command} );
+}
+
+sub check_group {
+	my ( $to ) = @_;
+
+	# Support comma-separated groups, like ONLINE,LEADERS. All checks must match.
+	return !grep { !check_group( $_ ) } split /\s*,\s*/, $to if index( $to, ',' ) > -1;
+
+	return 1 if $to eq 'ALL';
+	return 1 if $to eq 'AI=auto' && AI::state == AI::AUTO;
+	return 1 if $to eq 'AI=manual' && AI::state == AI::MANUAL;
+	return 1 if $to eq 'AI=off' && AI::state == AI::OFF;
+	return 1 if $to eq 'ONLINE' && $net->getState == Network::IN_GAME;
+	return 1 if $to eq 'OFFLINE' && $net->getState != Network::IN_GAME;
+	return 1 if $to eq 'LEADERS' && !$config{follow};
+	return 1 if $to eq 'FOLLOWERS' && $config{follow} && $config{followTarget};
+	return 1 if existsInList( $config{bus_command_groups}, $to );
+	return 1 if $to =~ /^MAP=(.*)$/o && $field && $1 eq $field->baseName;
+	return 1 if $to =~ /^(\w+)=(.*)$/o && $config{$1} eq $2;
+
+	return;
+}
+
+1;

plugins/bus_hook/bus_hook.pl

@@ -0,0 +1,63 @@
+package OpenKore::Plugins::BusHook;
+###############################################################################
+# Plugin to connect to a Bus server and proxy Bus messages through hooks.
+
+use strict;
+
+use Bus::Client;
+use Log qw( &debug );
+
+our $name = 'bus_hook';
+our $bus;
+
+Plugins::register( $name, "$name plugin", \&Unload, \&Unload );
+
+my $hooks = Plugins::addHooks(    #
+	[ 'start3'       => \&onStart ],
+	[ 'mainLoop_pre' => \&onMainLoopPre ],
+	[ 'bus/send'     => \&onBusSend ],
+);
+
+sub Unload {
+	$bus = undef;
+	Plugins::delHooks( $hooks );
+}
+
+sub onStart {
+	return if $bus;
+
+	$bus = $Globals::bus || Bus::Client->new( userAgent => "$name plugin" );
+	$bus->onConnected->add( undef, \&onConnected );
+	$bus->onMessageReceived->add( undef, \&onMessageReceived );
+}
+
+sub onMainLoopPre {
+	onStart() if !$bus;
+	$bus->iterate;
+}
+
+sub onBusSend {
+	my ( undef, $msg ) = @_;
+	Plugins::callHook( 'bus/sent' => $msg );
+	debug "[$name] >> sent $msg->{messageID}\n", $name;
+	$bus->send( $msg->{messageID}, $msg->{args} );
+}
+
+# Ask for a list of other clients.
+sub onConnected {
+	debug "[$name] >> connected\n", $name;
+	onBusSend( undef, { messageID => 'LIST_CLIENTS2' } );
+	Plugins::callHook( 'bus/connect' );
+}
+
+sub onMessageReceived {
+	my ( undef, undef, $msg ) = @_;
+
+	my $mid  = $msg->{messageID};
+	my $args = $msg->{args};
+	debug "[$name] << received $mid\n", $name;
+	Plugins::callHook( 'bus/recv'      => $msg );
+	Plugins::callHook( "bus/recv/$mid" => $args );
+}
+
+1;

plugins/bus_party/bus_party.pl

@@ -0,0 +1,145 @@
+package OpenKore::Plugins::BusParty;
+###############################################################################
+# Plugin to update party information via bus.
+
+use strict;
+
+use Globals qw( $accountID $char %config $field $net @partyUsersID );
+use Log qw( &debug &message );
+use Utils qw( &binAdd &binFind &binRemove &calcPosition &timeOut );
+use Time::HiRes qw( &time );
+
+our $name = 'bus_party';
+our $timeout = { time => 0, timeout => 0.2 };
+our $bus_party ||= {};
+
+Plugins::register( $name, "$name plugin", \&Unload, \&Unload );
+
+my $hooks = Plugins::addHooks(    #
+	[ 'mainLoop_pre'           => \&onMainLoopPre ],
+	[ 'bus/connect'            => \&onBusConnect ],
+	[ 'bus/recv/PARTY_REQUEST' => \&onBusRecvPartyRequest ],
+	[ 'bus/recv/PARTY_UPDATE'  => \&onBusRecvPartyUpdate ],
+	[ 'bus/recv/LEAVE'         => \&onBusRecvLeave ],
+);
+
+sub Unload {
+	Plugins::delHooks( $hooks );
+}
+
+sub onMainLoopPre {
+    return if !timeOut($timeout);
+    $timeout->{time} = time;
+
+	my $party_update = partial_party_update();
+	return if !%$party_update;
+	Plugins::callHook( 'bus/send', { messageID => 'PARTY_UPDATE', args => $party_update } );
+}
+
+sub onBusConnect{
+	Plugins::callHook( 'bus/send', { messageID => 'PARTY_REQUEST' } );
+	Plugins::callHook( 'bus/send', { messageID => 'PARTY_UPDATE', args => full_party_update() } );
+}
+
+sub onBusRecvPartyRequest {
+	my ( undef, $args ) = @_;
+	Plugins::callHook( 'bus/send', { messageID => 'PARTY_UPDATE', TO => $args->{FROM}, args => full_party_update() } );
+}
+
+sub onBusRecvPartyUpdate {
+	my ( undef, $args ) = @_;
+	my $actor = $bus_party->{ $args->{FROM} } ||= {};
+	$actor->{$_} = $args->{$_} foreach keys %$args;
+
+	# Update the party.
+	my $id = pack 'V', $actor->{id};
+	return if !$actor->{id} || $id eq $accountID;
+	return if !$actor->{name};
+	return if !$char;
+	return if !$char->{party};
+
+	my $party_user = $char->{party}{users}{$id};
+	if ( binFind( \@partyUsersID, $id ) eq '' ) {
+		binAdd( \@partyUsersID, $id );
+	}
+	if ( !$party_user ) {
+		$party_user = $char->{party}{users}{$id} = Actor::Party->new;
+		$party_user->{ID} = $id;
+		message "[bot_party] Party Member: $args->{name}\n";
+	}
+
+	foreach ( qw( name online hp hp_max ) ) {
+		$party_user->{$_} = $actor->{$_};
+	}
+	$party_user->{bus_party} = $actor->{party};
+	$party_user->{map}       = "$actor->{map}.gat";
+	$party_user->{pos}->{x}  = $actor->{x};
+	$party_user->{pos}->{y}  = $actor->{y};
+}
+
+sub onBusRecvLeave {
+	my ( undef, $args ) = @_;
+
+    my $actor = $bus_party->{ $args->{clientID} };
+    return if !$actor;
+
+	# Remove the character from $char->{party} if they're not in our actual party.
+	my $id = pack 'V', $actor->{id};
+	if ( $char && $char->{party} && ( !$char->{party}->{name} || $actor->{party} ne $char->{party}->{name} ) && binFind( \@partyUsersID, $id ) ne '' ) {
+		delete $char->{party}->{users}->{$id};
+		binRemove( \@partyUsersID, $id );
+	}
+
+	delete $bus_party->{ $args->{clientID} };
+}
+
+sub full_party_update {
+	my $update = {};
+	$update->{followTarget} = $config{follow} && $config{followTarget} || '';
+	if ( $char ) {
+		my $pos = calcPosition( $char );
+		$update->{id}         = unpack 'V', $accountID;
+		$update->{name}       = $char->{name};
+		$update->{hp}         = $char->{hp};
+		$update->{hp_max}     = $char->{hp_max};
+		$update->{sp}         = $char->{sp};
+		$update->{sp_max}     = $char->{sp_max};
+		$update->{lv}         = $char->{lv};
+		$update->{xp}         = $char->{exp};
+		$update->{xp_max}     = $char->{exp_max};
+		$update->{jl}         = $char->{lv_job};
+		$update->{jp}         = $char->{exp_job};
+		$update->{jp_max}     = $char->{exp_job_max};
+		$update->{zeny}       = $char->{zeny};
+		$update->{status}     = $char->{statuses} && %{ $char->{statuses} } ? join ', ', keys %{ $char->{statuses} } : '';
+		$update->{x}          = $pos->{x};
+		$update->{y}          = $pos->{y};
+		$update->{weight}     = $char->{weight};
+		$update->{weight_max} = $char->{weight_max};
+		if ( $char->{party} ) {
+			$update->{party} = $char->{party}->{name} || '';
+			$update->{admin} = $char->{party}->{users}->{$accountID}->{admin} ? 1 : 0;
+		}
+	}
+	if ( $field ) {
+		$update->{map} = $field->baseName;
+	}
+	if ( $net ) {
+		$update->{online} = $net->getState == Network::IN_GAME ? 1 : 0;
+	}
+	$update;
+}
+
+sub partial_party_update {
+	our $last_update ||= {};
+	my $update  = full_party_update();
+	my $partial = {};
+	foreach ( keys %$update ) {
+		next if $last_update->{$_} eq $update->{$_};
+		$partial->{$_} = $update->{$_};
+	}
+	$last_update = $update;
+	$partial;
+}
+
+1;

src/Bus/Client.pm

@@ -68,6 +68,7 @@ sub new {
 	my %args = @_;
 	my $self = bless {}, $class;
 
+	$self->{verbose} = $args{verbose};
 	$self->{host} = $args{host};
 	$self->{port} = $args{port};
 	$self->{userAgent}   = $args{userAgent} || "OpenKore";
@@ -80,6 +81,7 @@ sub new {
 	# connected to the bus.
 	$self->{sendQueue} = [];
 	$self->{seq} = 0;
+	$self->{onConnected} = new CallbackList();
 	$self->{onMessageReceived} = new CallbackList();
 	$self->{onDialogRequested} = new CallbackList();
 
@@ -104,38 +106,39 @@ sub iterate {
 
 	} elsif ($state == STARTING_SERVER) {
 		if (time - $self->{startTime} > RESTART_INTERVAL) {
-			#print "Starting\n";
+			print "Starting\n" if $self->{verbose};
 			my $starter = $self->{starter};
 			my $state = $starter->iterate();
 			if ($state == Bus::Server::Starter::STARTED) {
 				$self->{state} = HANDSHAKING;
 				$self->{host}  = $starter->getHost();
 				$self->{port}  = $starter->getPort();
-				#print "Bus server started at $self->{host}:$self->{port}\n";
+				print "Bus server started at $self->{host}:$self->{port}\n" if $self->{verbose};
 				$self->reconnect();
 				$self->{startTime} = time;
 
 			} elsif ($state == Bus::Server::Starter::FAILED) {
 				# Cannot start; try again.
-				#print "Start failed.\n";
+				print "Start failed.\n" if $self->{verbose};
 				$self->{starter} = new Bus::Server::Starter();
 				$self->{startTime} = time;
 			}
 		}
 
 	} elsif ($state == HANDSHAKING) {
-		#print "Handshaking\n";
+		print "Handshaking\n" if $self->{verbose};
 		my $ID;
 		my $args = $self->readNext(\$ID);
 		if ($args) {
-			#print "Sending HELLO\n";
+			print "Sending HELLO\n" if $self->{verbose};
 			$self->{ID} = $args->{yourID};
 			$self->{client}->send("HELLO", {
 				userAgent   => $self->{userAgent},
 				privateOnly => $self->{privateOnly}
 			});
 			$self->{state} = CONNECTED;
-			#print "Connected\n";
+			print "Connected\n" if $self->{verbose};
+			$self->onConnected->call($self);
 		}
 
 	} elsif ($state == CONNECTED) {
@@ -201,12 +204,12 @@ sub ID {
 sub reconnect {
 	my ($self) = @_;
 	eval {
-		#print "(Re)connecting\n";
+		print "(Re)connecting\n" if $self->{verbose};
 		$self->{client} = new Bus::SimpleClient($self->{host}, $self->{port});
 		$self->{state} = HANDSHAKING;
 	};
 	if (caught('SocketException')) {
-		#print "Cannot connect: $@\n";
+		print "Cannot connect: $@\n" if $self->{verbose};
 		$self->{state} = NOT_CONNECTED;
 		$self->{connectTime} = time;
 	} elsif ($@) {
@@ -243,7 +246,7 @@ sub readNext {
 		$args = $self->{client}->readNext($MID);
 	};
 	if (caught('IOException')) {
-		#print "Disconnected from IPC server.\n";
+		print "Disconnected from IPC server.\n" if $self->{verbose};
 		$self->handleIOException();
 		return undef;
 	} elsif ($@) {
@@ -273,6 +276,7 @@ sub send {
 			$self->{client}->send($MID, $args);
 		};
 		if (caught('IOException')) {
+			print "Failed to send $MID: " . $self->handleIOException . "\n";
 			$self->handleIOException();
 			push @{$self->{sendQueue}}, [$MID, $args];
 			return 0;
@@ -396,6 +400,16 @@ sub onMessageReceived {
 	return $_[0]->{onMessageReceived};
 }
 
+##
+# CallbackList $Bus_Client->onConnected()
+#
+# This event is triggered when the client connects.  It may happen more than
+# once, if the client has to reconnect for any reason.  The event argument
+# is undef.
+sub onConnected {
+	return $_[0]->{onConnected};
+}
+
 sub onDialogRequested {
 	return $_[0]->{onDialogRequested};
 }

src/Bus/MessageParser.pm

@@ -20,7 +20,7 @@ sub readNext {
 	my ($self, $ID) = @_;
 	my $processed;
 	my $args = unserialize($self->{buffer}, $ID, \$processed);
-	if ($args) {
+	if ($$ID) {
 		substr($self->{buffer}, 0, $processed, '');
 	}
 	return $args;