LP#1350457: Pass caller's session to subrequests called via method_lookup
[OpenSRF.git] / src / perl / lib / OpenSRF / Application.pm
1 package OpenSRF::Application;
2 # vim:noet:ts=4
3 use vars qw/$_app $log @_METHODS $thunk $server_class/;
4
5 use base qw/OpenSRF/;
6 use OpenSRF::AppSession;
7 use OpenSRF::DomainObject::oilsMethod;
8 use OpenSRF::DomainObject::oilsResponse qw/:status/;
9 use OpenSRF::Utils::Logger qw/:level $logger/;
10 use Data::Dumper;
11 use Time::HiRes qw/time/;
12 use OpenSRF::EX qw/:try/;
13 use Carp;
14 use OpenSRF::Utils::JSON;
15
16 sub DESTROY{};
17
18 use strict;
19 use warnings;
20
21 $log = 'OpenSRF::Utils::Logger';
22
23 our $in_request = 0;
24 our @pending_requests;
25 our $shared_conf;
26
27 sub package {
28         my $self = shift;
29         return 1 unless ref($self);
30         return $self->{package};
31 }
32
33 sub signature {
34         my $self = shift;
35         return 0 unless ref($self);
36         return $self->{signature};
37 }
38
39 sub strict {
40     my $self = shift; 
41     return 0 unless ref($self);
42     return $self->{strict};
43 }
44
45 sub argc {
46         my $self = shift;
47         return 0 unless ref($self);
48         return $self->{argc};
49 }
50
51 sub max_chunk_size {
52         my $self = shift;
53         return 0 unless ref($self);
54         return $self->{max_chunk_size} if (defined($self->{max_chunk_size}));
55         return 10240;
56 }
57
58 sub max_chunk_count {
59         my $self = shift;
60         return 0 unless ref($self);
61         return $self->{max_chunk_count} || 0;
62 }
63
64 sub api_name {
65         my $self = shift;
66         return 1 unless ref($self);
67         return $self->{api_name};
68 }
69
70 sub api_level {
71         my $self = shift;
72         return 1 unless ref($self);
73         return $self->{api_level};
74 }
75
76 sub session {
77         my $self = shift;
78         my $session = shift;
79
80         if($session) {
81                 $self->{session} = $session;
82         }
83         return $self->{session};
84 }
85
86 sub server_class {
87         my $class = shift;
88         if($class) {
89                 $server_class = $class;
90         }
91         return $server_class;
92 }
93
94 sub thunk {
95         my $self = shift;
96         my $flag = shift;
97         $thunk = $flag if (defined $flag);
98         return $thunk;
99 }
100
101 sub application_implementation {
102         my $self = shift;
103         my $app = shift;
104
105         if (defined $app) {
106                 $_app = $app;
107                 $_app->use;
108                 if( $@ ) {
109                         $log->error( "Error loading application_implementation: $app -> $@", ERROR);
110                 }
111
112         }
113
114         return $_app;
115 }
116
117 sub handler {
118         my ($self, $session, $app_msg) = @_;
119
120         if( ! $app_msg ) {
121                 return 1;  # error?
122         }
123
124         my $app = $self->application_implementation;
125
126         if ($session->last_message_type eq 'REQUEST') {
127
128         my @p = $app_msg->params;
129                 my $method_name = $app_msg->method;
130                 my $method_proto = $session->last_message_api_level;
131
132                 # By default, we log all method params at the info level
133                 # Here we are consult our shared portion of the config file
134                 # to look for any exceptions to this behavior
135                 my $logdata = "CALL: ".$session->service." $method_name ";
136                 my $redact_params = 0;
137                 if (@p) {
138                         if (ref($shared_conf->shared->log_protect) eq 'ARRAY') {
139                                 foreach my $match_string (@{$shared_conf->shared->log_protect}) {
140                                         if ($method_name =~ /^$match_string/) {
141                                                 $redact_params = 1;
142                                                 last;
143                                         }
144                                 }
145                         }
146                         if ($redact_params) {
147                                 $logdata .= "**PARAMS REDACTED**";
148                         } else {
149                                 $logdata .= join(', ', map { (defined $_) ? $_ : '' } @p);
150                         }
151                 }
152                 $log->info($logdata);
153
154                 my $coderef = $app->method_lookup( $method_name, $method_proto, 1, 1 );
155
156                 unless ($coderef) {
157                         $session->status( OpenSRF::DomainObject::oilsMethodException->new( 
158                                                 statusCode => STATUS_NOTFOUND(),
159                                                 status => "Method [$method_name] not found for $app"));
160                         return 1;
161                 }
162
163                 unless ($session->continue_request) {
164                         $session->status(
165                                 OpenSRF::DomainObject::oilsConnectStatus->new(
166                                                 statusCode => STATUS_REDIRECTED(),
167                                                 status => 'Disconnect on max requests' ) );
168                         $session->kill_me;
169                         return 1;
170                 }
171
172                 if (ref $coderef) {
173                         my @args = $app_msg->params;
174                         $coderef->session( $session );
175                         my $appreq = OpenSRF::AppRequest->new( $session );
176                         $appreq->max_chunk_size( $coderef->max_chunk_size );
177                         $appreq->max_chunk_count( $coderef->max_chunk_count );
178
179                         $log->debug( "in_request = $in_request : [" . $appreq->threadTrace."]", INTERNAL );
180                         if( $in_request ) {
181                                 $log->debug( "Pushing onto pending requests: " . $appreq->threadTrace, DEBUG );
182                                 push @pending_requests, [ $appreq, \@args, $coderef ]; 
183                                 return 1;
184                         }
185
186
187                         $in_request++;
188
189                         $log->debug( "Executing coderef for {$method_name}", INTERNAL );
190
191                         my $resp;
192                         try {
193                                 # un-if(0) this block to enable param checking based on signature and argc
194                                 if ($coderef->strict) {
195                                         if (@args < $coderef->argc) {
196                                                 die     "Not enough params passed to ".
197                                                         $coderef->api_name." : requires ". $coderef->argc
198                                         }
199                                         if (@args) {
200                                                 my $sig = $coderef->signature;
201                                                 if ($sig && exists $sig->{params}) {
202                                                         for my $p (0 .. scalar(@{ $sig->{params} }) - 1 ) {
203                                                                 my $s = $sig->{params}->[$p];
204                                                                 my $a = $args[$p];
205                                                                 if ($s->{class} && OpenSRF::Utils::JSON->lookup_hint(ref $a) ne $s->{class}) {
206                                                                         die "Incorrect param class at position $p : should be a '$$s{class}'";
207                                                                 } elsif ($s->{type}) {
208                                                                         if (lc($s->{type}) eq 'object' && $a !~ /HASH/o) {
209                                                                                 die "Incorrect param type at position $p : should be an 'object'";
210                                                                         } elsif (lc($s->{type}) eq 'array' && $a !~ /ARRAY/o) {
211                                                                                 die "Incorrect param type at position $p : should be an 'array'";
212                                                                         } elsif (lc($s->{type}) eq 'number' && (ref($a) || $a !~ /^-?\d+(?:\.\d+)?$/o)) {
213                                                                                 die "Incorrect param type at position $p : should be a 'number'";
214                                                                         } elsif (lc($s->{type}) eq 'string' && ref($a)) {
215                                                                                 die "Incorrect param type at position $p : should be a 'string'";
216                                                                         }
217                                                                 }
218                                                         }
219                                                 }
220                                         }
221                                 }
222
223                                 my $start = time();
224                                 $resp = $coderef->run( $appreq, @args); 
225                                 my $time = sprintf '%.3f', time() - $start;
226
227                                 $log->debug( "Method duration for [$method_name]:  ". $time );
228                                 $appreq->respond_complete( $resp );
229
230                         } catch Error with {
231                                 my $e = shift;
232                                 warn "Caught error from 'run' method: $e\n";
233
234                                 if(UNIVERSAL::isa($e,"Error")) {
235                                         $e = $e->stringify();
236                                 } 
237                                 my $sess_id = $session->session_id;
238                                 $session->status(
239                                         OpenSRF::DomainObject::oilsMethodException->new(
240                                                         statusCode      => STATUS_INTERNALSERVERERROR(),
241                                                         status          => " *** Call to [$method_name] failed for session ".
242                                                                            "[$sess_id], thread trace ".
243                                                                            "[".$appreq->threadTrace."]:\n$e\n"
244                                         )
245                                 );
246                         };
247
248
249
250                         # ----------------------------------------------
251
252
253                         # XXX may need this later
254                         # $_->[1] = 1 for (@OpenSRF::AppSession::_CLIENT_CACHE);
255
256                         $in_request--;
257
258                         $log->debug( "Pending Requests: " . scalar(@pending_requests), INTERNAL );
259
260                         # cycle through queued requests
261                         while( my $aref = shift @pending_requests ) {
262                                 $in_request++;
263                                 my $resp;
264                                 try {
265                                         # un-if(0) this block to enable param checking based on signature and argc
266                                         if (0) {
267                                                 if (@args < $aref->[2]->argc) {
268                                                         die     "Not enough params passed to ".
269                                                                 $aref->[2]->api_name." : requires ". $aref->[2]->argc
270                                                 }
271                                                 if (@args) {
272                                                         my $sig = $aref->[2]->signature;
273                                                         if ($sig && exists $sig->{params}) {
274                                                                 for my $p (0 .. scalar(@{ $sig->{params} }) - 1 ) {
275                                                                         my $s = $sig->{params}->[$p];
276                                                                         my $a = $args[$p];
277                                                                         if ($s->{class} && OpenSRF::Utils::JSON->lookup_hint(ref $a) ne $s->{class}) {
278                                                                                 die "Incorrect param class at position $p : should be a '$$s{class}'";
279                                                                         } elsif ($s->{type}) {
280                                                                                 if (lc($s->{type}) eq 'object' && $a !~ /HASH/o) {
281                                                                                         die "Incorrect param type at position $p : should be an 'object'";
282                                                                                 } elsif (lc($s->{type}) eq 'array' && $a !~ /ARRAY/o) {
283                                                                                         die "Incorrect param type at position $p : should be an 'array'";
284                                                                                 } elsif (lc($s->{type}) eq 'number' && (ref($a) || $a !~ /^-?\d+(?:\.\d+)?$/o)) {
285                                                                                         die "Incorrect param type at position $p : should be a 'number'";
286                                                                                 } elsif (lc($s->{type}) eq 'string' && ref($a)) {
287                                                                                         die "Incorrect param type at position $p : should be a 'string'";
288                                                                                 }
289                                                                         }
290                                                                 }
291                                                         }
292                                                 }
293                                         }
294
295                                         my $start = time;
296                                         my $response = $aref->[2]->run( $aref->[0], @{$aref->[1]} );
297                                         my $time = sprintf '%.3f', time - $start;
298                                         $log->debug( "Method duration for [".$aref->[2]->api_name." -> ".join(', ',@{$aref->[1]}).']:  '.$time, DEBUG );
299
300                                         $appreq = $aref->[0];   
301                                         $appreq->respond_complete( $response );
302                                         $log->debug( "Executed: " . $appreq->threadTrace, INTERNAL );
303
304                                 } catch Error with {
305                                         my $e = shift;
306                                         if(UNIVERSAL::isa($e,"Error")) {
307                                                 $e = $e->stringify();
308                                         }
309                                         $session->status(
310                                                 OpenSRF::DomainObject::oilsMethodException->new(
311                                                                 statusCode => STATUS_INTERNALSERVERERROR(),
312                                                                 status => "Call to [".$aref->[2]->api_name."] faild:  $e"
313                                                 )
314                                         );
315                                 };
316                                 $in_request--;
317                         }
318
319                         return 1;
320                 } 
321
322                 $log->info("Received non-REQUEST message in Application handler");
323
324                 my $res = OpenSRF::DomainObject::oilsMethodException->new( 
325                                 status => "Received non-REQUEST message in Application handler");
326                 $session->send('ERROR', $res);
327                 $session->kill_me;
328                 return 1;
329
330         } else {
331                 $session->push_queue([ $app_msg, $session->last_threadTrace ]);
332         }
333
334         $session->last_message_type('');
335         $session->last_message_api_level('');
336
337         return 1;
338 }
339
340 sub is_registered {
341         my $self = shift;
342         my $api_name = shift;
343         my $api_level = shift || 1;
344         return exists($_METHODS[$api_level]{$api_name});
345 }
346
347
348 sub normalize_whitespace {
349         my $txt = shift;
350
351         $txt =~ s/^\s+//gso;
352         $txt =~ s/\s+$//gso;
353         $txt =~ s/\s+/ /gso;
354         $txt =~ s/\n//gso;
355         $txt =~ s/\. /\.  /gso;
356
357         return $txt;
358 }
359
360 sub parse_string_signature {
361         my $string = shift;
362         return [] unless $string;
363         my @chunks = split(/\@/smo, $string);
364
365         my @params;
366         my $ret;
367         my $desc = '';
368         for (@chunks) {
369                 if (/^return (.+)$/so) {
370                         $ret = [normalize_whitespace($1)];
371                 } elsif (/^param (\w+) \b(.+)$/so) {
372                         push @params, [ $1, normalize_whitespace($2) ];
373                 } else {
374                         $desc .= '@' if $desc;
375                         $desc .= $_;
376                 }
377         }
378
379         return [normalize_whitespace($desc),\@params, $ret];
380 }
381
382 sub parse_array_signature {
383         my $array = shift;
384         my ($d,$p,$r) = @$array;
385         return {} unless ($d or $p or $r);
386
387         return {
388                 desc    => $d,
389                 params  => [
390                         map { 
391                                 { name  => $$_[0],
392                                   desc  => $$_[1],
393                                   type  => $$_[2],
394                                   class => $$_[3],
395                                 }
396                         } @$p
397                 ],
398                 'return'=>
399                         { desc  => $$r[0],
400                           type  => $$r[1],
401                           class => $$r[2],
402                         }
403         };
404 }
405
406 sub register_method {
407         my $self = shift;
408         my $app = ref($self) || $self;
409         my %args = @_;
410
411
412         throw OpenSRF::DomainObject::oilsMethodException unless ($args{method});
413
414         $args{api_level} = 1 unless(defined($args{api_level}));
415         $args{stream} ||= 0;
416         $args{remote} ||= 0;
417         $args{argc} ||= 0;
418         $args{package} ||= $app;                
419         $args{server_class} = server_class();
420         $args{api_name} ||= $args{server_class} . '.' . $args{method};
421
422         # un-if(0) this block to enable signature parsing
423         if (!$args{signature}) {
424                 if ($args{notes} && !ref($args{notes})) {
425                         $args{signature} =
426                                 parse_array_signature( parse_string_signature( $args{notes} ) );
427                 }
428         } elsif( !ref($args{signature}) ) {
429                 $args{signature} =
430                         parse_array_signature( parse_string_signature( $args{signature} ) );
431         } elsif( ref($args{signature}) eq 'ARRAY') {
432                 $args{signature} =
433                         parse_array_signature( $args{signature} );
434         }
435         
436         unless ($args{object_hint}) {
437                 ($args{object_hint} = $args{package}) =~ s/::/_/go;
438         }
439
440         OpenSRF::Utils::JSON->register_class_hint(
441                 strip => ['session'],
442                 name => $app,
443                 hint => $args{object_hint},
444                 type => "hash"
445         );
446
447         $_METHODS[$args{api_level}]{$args{api_name}} = bless \%args => $app;
448
449         __PACKAGE__->register_method(
450                 stream => 0,
451                 argc => $args{argc},
452                 api_name => $args{api_name}.'.atomic',
453                 method => 'make_stream_atomic',
454                 notes => "This is a system generated method.  Please see the definition for $args{api_name}",
455         ) if ($args{stream});
456 }
457
458 sub retrieve_remote_apis {
459         my $method = shift;
460         my $session = OpenSRF::AppSession->create('router');
461         try {
462                 $session->connect or OpenSRF::EX::WARN->throw("Connection to router timed out");
463         } catch Error with {
464                 my $e = shift;
465                 $log->debug( "Remote subrequest returned an error:\n". $e );
466                 return undef;
467         } finally {
468                 return undef unless ($session->state == $session->CONNECTED);
469         };
470
471         my $req = $session->request( 'opensrf.router.info.class.list' );
472         my $list = $req->recv;
473
474         if( UNIVERSAL::isa($list,"Error") ) {
475                 throw $list;
476         }
477
478         my $content = $list->content;
479
480         $req->finish;
481         $session->finish;
482         $session->disconnect;
483
484         my %u_list = map { ($_ => 1) } @$content;
485
486         for my $class ( keys %u_list ) {
487                 next if($class eq $server_class);
488                 populate_remote_method_cache($class, $method);
489         }
490 }
491
492 sub populate_remote_method_cache {
493         my $class = shift;
494         my $meth = shift;
495
496         my $session = OpenSRF::AppSession->create($class);
497         try {
498                 $session->connect or OpenSRF::EX::WARN->throw("Connection to $class timed out");
499
500                 my $call = 'opensrf.system.method.all' unless (defined $meth);
501                 $call = 'opensrf.system.method' if (defined $meth);
502
503                 my $req = $session->request( $call, $meth );
504
505                 while (my $method = $req->recv) {
506                         next if (UNIVERSAL::isa($method, 'Error'));
507
508                         $method = $method->content;
509                         next if ( exists($_METHODS[$$method{api_level}]) &&
510                                 exists($_METHODS[$$method{api_level}]{$$method{api_name}}) );
511                         $method->{remote} = 1;
512                         bless($method, __PACKAGE__ );
513                         $_METHODS[$$method{api_level}]{$$method{api_name}} = $method;
514                 }
515
516                 $req->finish;
517                 $session->finish;
518                 $session->disconnect;
519
520         } catch Error with {
521                 my $e = shift;
522                 $log->debug( "Remote subrequest returned an error:\n". $e );
523                 return undef;
524         };
525 }
526
527 sub method_lookup {             
528         my $self = shift;
529         my $method = shift;
530         my $proto = shift;
531         my $no_recurse = shift || 0;
532         my $no_remote = shift || 0;
533
534         # this instead of " || 1;" above to allow api_level 0
535         $proto = $self->api_level unless (defined $proto);
536
537         my $class = ref($self) || $self;
538
539         $log->debug("Lookup of [$method] by [$class] in api_level [$proto]", DEBUG);
540         $log->debug("Available methods\n\t".join("\n\t", keys %{ $_METHODS[$proto] }), INTERNAL);
541
542         my $meth;
543         if (__PACKAGE__->thunk) {
544                 for my $p ( reverse(1 .. $proto) ) {
545                         if (exists $_METHODS[$p]{$method}) {
546                                 $meth = $_METHODS[$p]{$method};
547                         }
548                 }
549         } else {
550                 if (exists $_METHODS[$proto]{$method}) {
551                         $meth = $_METHODS[$proto]{$method};
552                 }
553         }
554
555         if (defined $meth) {
556                 if($no_remote and $meth->{remote}) {
557                         $log->debug("OH CRAP We're not supposed to return remote methods", WARN);
558                         return undef;
559                 }
560
561         } elsif (!$no_recurse) {
562                 $log->debug("We didn't find [$method], asking everyone else.", DEBUG);
563                 retrieve_remote_apis($method);
564                 $meth = $self->method_lookup($method,$proto,1);
565         }
566
567         $meth->session($self->session) if $meth && ref($self); # Pass the caller's session
568         return $meth;
569 }
570
571 sub run {
572         my $self = shift;
573         my $req = shift;
574
575         my $resp;
576         my @params = @_;
577
578         if ( !UNIVERSAL::isa($req, 'OpenSRF::AppRequest') ) {
579                 $log->debug("Creating a SubRequest object", DEBUG);
580                 unshift @params, $req;
581                 $req = OpenSRF::AppSubrequest->new( session => $self->session );
582         } else {
583                 $log->debug("This is a top level request", DEBUG);
584         }
585
586         if (!$self->{remote}) {
587                 my $code = \&{$self->{package} . '::' . $self->{method}};
588                 my $err = undef;
589                 my $warnhandler;
590
591                 try {
592                         $warnhandler = $SIG{__WARN__};
593                         $SIG{__WARN__} = sub {
594                                 (my $msg = shift) =~ s/\n$//;
595                                 $log->warn($self->{api_name} . ": $msg");
596                                 return 1; # prevents warning going out to stderr
597                         };
598
599                         $resp = $code->($self, $req, @params);
600
601                 } catch Error with {
602                         $err = shift;
603
604                         if( ref($self) eq 'HASH') {
605                                 $log->error("Sub $$self{package}::$$self{method} DIED!!!\n\t$err\n", ERROR);
606                         }
607                 } finally {
608                         $SIG{__WARN__} = $warnhandler;
609                 };
610
611                 if($err) {
612                         if(UNIVERSAL::isa($err,"Error")) { 
613                                 throw $err;
614                         } else {
615                                 die $err->stringify; 
616                         }
617                 }
618
619
620                 $log->debug("Coderef for [$$self{package}::$$self{method}] has been run", DEBUG);
621
622                 if ( ref($req) and UNIVERSAL::isa($req, 'OpenSRF::AppSubrequest') ) {
623                         $req->respond($resp) if (defined $resp);
624                         $log->debug("SubRequest object is responding with : " . join(" ",$req->responses), DEBUG);
625                         return $req->responses;
626                 } else {
627                         $log->debug("A top level Request object is responding $resp", DEBUG) if (defined $resp);
628                         return $resp;
629                 }
630         } else {
631                 my $session = OpenSRF::AppSession->create($self->{server_class});
632                 try {
633                         #$session->connect or OpenSRF::EX::WARN->throw("Connection to [$$self{server_class}] timed out");
634                         my $remote_req = $session->request( $self->{api_name}, @params );
635                         while (my $remote_resp = $remote_req->recv) {
636                                 OpenSRF::Utils::Logger->debug("Remote Subrequest Received " . $remote_resp, INTERNAL );
637                                 if( UNIVERSAL::isa($remote_resp,"Error") ) {
638                                         throw $remote_resp;
639                                 }
640                                 $req->respond( $remote_resp->content );
641                         }
642                         $remote_req->finish();
643
644                 } catch Error with {
645                         my $e = shift;
646                         $log->debug( "Remote subrequest returned an error:\n". $e );
647                         return undef;
648                 };
649
650                 if ($session) {
651                         $session->disconnect();
652                         $session->finish();
653                 }
654
655                 $log->debug( "Remote Subrequest Responses " . join(" ", $req->responses), INTERNAL );
656
657                 return $req->responses;
658         }
659         # huh? how'd we get here...
660         return undef;
661 }
662
663 sub introspect {
664         my $self = shift;
665         my $client = shift;
666         my $method = shift;
667         my $limit = shift;
668         my $offset = shift;
669
670         if ($self->api_name =~ /all$/o) {
671                 $offset = $limit;
672                 $limit = $method;
673                 $method = undef; 
674         }
675
676         my ($seen,$returned) = (0,0);
677         for my $api_level ( reverse(1 .. $#_METHODS) ) {
678                 for my $api_name ( sort keys %{$_METHODS[$api_level]} ) {
679                         if (!$offset || $offset <= $seen) {
680                                 if (!$_METHODS[$api_level]{$api_name}{remote}) {
681                                         if (defined($method)) {
682                                                 if ($api_name =~ $method) {
683                                                         if (!$limit || $returned < $limit) {
684                                                                 $client->respond( $_METHODS[$api_level]{$api_name} );
685                                                                 $returned++;
686                                                         }
687                                                 }
688                                         } else {
689                                                 if (!$limit || $returned < $limit) {
690                                                         $client->respond( $_METHODS[$api_level]{$api_name} );
691                                                         $returned++;
692                                                 }
693                                         }
694                                 }
695                         }
696                         $seen++;
697                 }
698         }
699
700         return undef;
701 }
702 __PACKAGE__->register_method(
703         stream => 1,
704         method => 'introspect',
705         api_name => 'opensrf.system.method.all',
706         argc => 0,
707         signature => {
708                 desc => q/This method is used to introspect an entire OpenSRF Application/,
709                 return => {
710                         desc => q/A stream of objects describing the methods available via this OpenSRF Application/,
711                         type => 'object'
712                 }
713         },
714 );
715 __PACKAGE__->register_method(
716         stream => 1,
717         method => 'introspect',
718         argc => 1,
719         api_name => 'opensrf.system.method',
720         argc => 1,
721         signature => {
722                 desc => q/Use this method to get the definition of a single OpenSRF Method/,
723                 params => [
724                         { desc => q/The method to introspect/,
725                           type => 'string' },
726                 ],
727                 return => { desc => q/An object describing the method requested, or an error if it can't be found/,
728                             type => 'object' }
729         },
730 );
731
732 sub echo_method {
733         my $self = shift;
734         my $client = shift;
735         my @args = @_;
736
737         $client->respond( $_ ) for (@args);
738         return undef;
739 }
740 __PACKAGE__->register_method(
741         stream => 1,
742         method => 'echo_method',
743         argc => 1,
744         api_name => 'opensrf.system.echo',
745         signature => {
746                 desc => q/A test method that will echo back it's arguments in a streaming response/,
747                 params => [
748                         { desc => q/One or more arguments to echo back/ }
749                 ],
750                 return => { desc => q/A stream of the arguments passed/ }
751         },
752 );
753
754 sub time_method {
755         my( $self, $conn ) = @_;
756         return CORE::time;
757 }
758 __PACKAGE__->register_method(
759         method => 'time_method',
760         argc => 0,
761         api_name => 'opensrf.system.time',
762         signature => {
763                 desc => q/Returns the current system time as epoch seconds/,
764                 return => { desc => q/epoch seconds/ }
765         }
766 );
767
768 sub make_stream_atomic {
769         my $self = shift;
770         my $req = shift;
771         my @args = @_;
772
773         (my $m_name = $self->api_name) =~ s/\.atomic$//o;
774         my $m = $self->method_lookup($m_name);
775
776         $m->session( $req->session );
777         my @results = $m->run(@args);
778         $m->session('');
779
780         return \@results;
781 }
782
783 1;