view release on metacpan or  search on metacpan

lib/Grpc/Client/BaseStub.pm  view on Meta::CPAN

## Base class for generated client stubs. Stub methods are expected to call
## _simpleRequest or _streamRequest and return the result.

use Grpc::XS;
use Grpc::XS::Channel;
use Grpc::XS::Timeval;

use Grpc::Constants;

use Grpc::Client::UnaryCall;
use Grpc::Client::ClientStreamingCall;
use Grpc::Client::ServerStreamingCall;
use Grpc::Client::BidiStreamingCall;

use constant true  => 1;
use constant false => 0;

## params:
##    - 'hostname': string
##    - 'update_metadata': (optional) a callback function which takes in a
##                         metadata array, and returns an updated metadata array
##    - 'grpc.primary_user_agent': (optional) a user-agent string

sub new {
	my $proto    = shift;
	my $hostname = shift;
	my %param		 = @_;
	my $update_metadata    = $param{"update_metadata"};
	my $primary_user_agent = $param{"grpc.primary_user_agent"};
	my $credentials        = $param{"credentials"};
	my $timeout            = $param{"timeout"};

	unless (defined($primary_user_agent)) {
		$primary_user_agent = "grpc-perl/".($Grpc::XS::VERSION);
	}
	
	if (!exists($param{"credentials"})) {
		die("The 'credentials' key is now required. Please see one of the ".
			  "Grpc::XS::ChannelCredentials::create methods");
	} elsif (!defined($param{"credentials"})) {
		delete $param{"credentials"};
	}

	my $channel = new Grpc::XS::Channel($hostname,%param);

	my $self = {
		'_hostname'           => $hostname,
		'_channel'            => undef,
		'_update_metadata'    => $update_metadata,	## a callback function
		'_primary_user_agent' => $primary_user_agent,
		'_credentials'        => $credentials,
		'_channel'            => $channel,
		'_timeout'            => $timeout,
	};
	bless $self,$proto;

	return $self;
}

## return string The URI of the endpoint.

sub getTarget {
	my $self = shift;
	return $self->{_channel}->getTarget();
}

## $try_to_connect bool
## return int The grpc connectivity state

sub getConnectivityState {
	my $self = shift;
	my $try_to_connect = shift||false;
	return $self->{_channel}->getConnectivityState($try_to_connect);
}

## $timeout in microseconds
## return bool true if channel is ready
## dies if channel is in FATAL_ERROR state

sub waitForReady {
	my $self = shift;
	my $timeout = shift;

    my $new_state = $self->getConnectivityState(true);
    if ($self->_checkConnectivityState($new_state)) {
    	return true;
	}

  my $now = Grpc::XS::Timeval::now();
 	my $delta = new Grpc::XS::Timeval($timeout);
  my $deadline = Grpc::XS::Timeval::add($now,$delta);

  while ($self->{_channel}->watchConnectivityState($new_state, $deadline)) {
		## state has changed before deadline
		$new_state = $self->getConnectivityState();
	  if ($self->_checkConnectivityState($new_state)) {
			return true;
	  }
	}

  ## deadline has passed
  $new_state = $self->getConnectivityState();

	return $self->_checkConnectivityState($new_state);
}

sub _checkConnectivityState {
	my $self = shift;
	my $new_state = shift;

	if ($new_state == Grpc::Constants::GRPC_CHANNEL_READY()){
    	return true;
	}
	if ($new_state == Grpc::Constants::GRPC_CHANNEL_SHUTDOWN()){
    	die('Failed to connect to server');
	}

	return false;
}