Subversion Repositories SmartDukaan

#

# Licensed to the Apache Software Foundation (ASF) under one

# or more contributor license agreements. See the NOTICE file

# distributed with this work for additional information

# regarding copyright ownership. The ASF licenses this file

# to you under the Apache License, Version 2.0 (the

# "License"); you may not use this file except in compliance

# with the License. You may obtain a copy of the License at

#

#   http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing,

# software distributed under the License is distributed on an

# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY

# KIND, either express or implied. See the License for the

# specific language governing permissions and limitations

# under the License.

#

require 5.6.0;

use strict;

use warnings;

use Thrift;

use Thrift::BufferedTransport;

use Thrift::BinaryProtocol;

#

# Server base class module

#

package Thrift::Server;

# 3 possible constructors:

#   1.  (processor, serverTransport)

#   2.  (processor, serverTransport, transportFactory, protocolFactory)

#   3.  (processor, serverTransport,

#        inputTransportFactory, outputTransportFactory,

#        inputProtocolFactory, outputProtocolFactory)

sub new

{

    my $classname    = shift;

    my @args         = @_;

    my $self;

    if (scalar @args == 2)

    {

      $self = _init($args[0], $args[1],

                    Thrift::BufferedTransportFactory->new(),

                    Thrift::BufferedTransportFactory->new(),

                    Thrift::BinaryProtocolFactory->new(),

                    Thrift::BinaryProtocolFactory->new());

    }

    elsif (scalar @args == 4)

    {

      $self = _init($args[0], $args[1], $args[2], $args[2], $args[3], $args[3]);

    }

    elsif (scalar @args == 6)

    {

      $self = _init($args[0], $args[1], $args[2], $args[3], $args[4], $args[5]);

    }

    else

    {

      die "Thrift::Server expects exactly 2, 4, or 6 args";

    }

    return bless($self,$classname);

}

sub _init

{

    my $processor              = shift;

    my $serverTransport        = shift;

    my $inputTransportFactory  = shift;

    my $outputTransportFactory = shift;

    my $inputProtocolFactory   = shift;

    my $outputProtocolFactory  = shift;

    my $self = {

        processor              => $processor,

        serverTransport        => $serverTransport,

        inputTransportFactory  => $inputTransportFactory,

        outputTransportFactory => $outputTransportFactory,

        inputProtocolFactory   => $inputProtocolFactory,

        outputProtocolFactory  => $outputProtocolFactory,

    };

}

sub serve

{

    die "abstract";

}

sub _clientBegin

{

    my $self  = shift;

    my $iprot = shift;

    my $oprot = shift;

    if (exists  $self->{serverEventHandler} and

        defined $self->{serverEventHandler})

    {

        $self->{serverEventHandler}->clientBegin($iprot, $oprot);

    }

}

sub _handleException

{

    my $self = shift;

    my $e    = shift;

    if ($e =~ m/TException/ and exists $e->{message}) {

        my $message = $e->{message};

        my $code    = $e->{code};

        my $out     = $code . ':' . $message;

        $message =~ m/TTransportException/ and die $out;

        if ($message =~ m/TSocket/) {

            # suppress TSocket messages

        } else {

            warn $out;

        }

    } else {

        warn $e;

    }

}

#

# SimpleServer from the Server base class that handles one connection at a time

#

package Thrift::SimpleServer;

use base qw( Thrift::Server );

sub new

{

    my $classname = shift;

    my @args      = @_;

    my $self      = $classname->SUPER::new(@args);

    return bless($self,$classname);

}

sub serve

{

    my $self = shift;

    $self->{serverTransport}->listen();

    while (1)

    {

        my $client = $self->{serverTransport}->accept();

        my $itrans = $self->{inputTransportFactory}->getTransport($client);

        my $otrans = $self->{outputTransportFactory}->getTransport($client);

        my $iprot  = $self->{inputProtocolFactory}->getProtocol($itrans);

        my $oprot  = $self->{outputProtocolFactory}->getProtocol($otrans);

        eval {

            $self->_clientBegin($iprot, $oprot);

            while (1)

            {

                $self->{processor}->process($iprot, $oprot);

            }

        }; if($@) {

            $self->_handleException($@);

        }

        $itrans->close();

        $otrans->close();

    }

}

#

# ForkingServer that forks a new process for each request

#

package Thrift::ForkingServer;

use base qw( Thrift::Server );

use POSIX ":sys_wait_h";

sub new

{

    my $classname = shift;

    my @args      = @_;

    my $self      = $classname->SUPER::new(@args);

    return bless($self,$classname);

}

sub serve

{

    my $self = shift;

    $self->{serverTransport}->listen();

    while (1)

    {

        my $client = $self->{serverTransport}->accept();

        $self->_client($client);

    }

}

sub _client

{

    my $self   = shift;

    my $client = shift;

    eval {

        my $itrans = $self->{inputTransportFactory}->getTransport($client);

        my $otrans = $self->{outputTransportFactory}->getTransport($client);

        my $iprot  = $self->{inputProtocolFactory}->getProtocol($itrans);

        my $oprot  = $self->{outputProtocolFactory}->getProtocol($otrans);

        $self->_clientBegin($iprot, $oprot);

        my $pid = fork();

        if ($pid) #parent

        {

            $self->_parent($pid, $itrans, $otrans);

        } else {

            $self->_child($itrans, $otrans, $iprot, $oprot);

        }

    }; if($@) {

        $self->_handleException($@);

    }

}

sub _parent

{

    my $self   = shift;

    my $pid    = shift;

    my $itrans = shift;

    my $otrans = shift;

    # add before collect, otherwise you race w/ waitpid

    $self->{children}->{$pid} = 1;

    $self->_collectChildren();

    # Parent must close socket or the connection may not get closed promptly

    $self->tryClose($itrans);

    $self->tryClose($otrans);

}

sub _child

{

    my $self   = shift;

    my $itrans = shift;

    my $otrans = shift;

    my $iprot  = shift;

    my $oprot  = shift;

    my $ecode = 0;

    eval {

        while (1)

        {

            $self->{processor}->process($iprot, $oprot);

        }

    }; if($@) {

        $ecode = 1;

        $self->_handleException($@);

    }

    $self->tryClose($itrans);

    $self->tryClose($otrans);

    exit($ecode);

}

sub tryClose

{

    my $self = shift;

    my $file = shift;

    eval {

        if (defined $file)

        {

          $file->close();

        }

    }; if($@) {

        if ($@ =~ m/TException/ and exists $@->{message}) {

            my $message = $@->{message};

            my $code    = $@->{code};

            my $out     = $code . ':' . $message;

            warn $out;

        } else {

            warn $@;

        }

    }

}

sub _collectChildren

{

    my $self = shift;

    while (scalar keys %{$self->{children}})

    {

        my $pid    = waitpid(-1, WNOHANG);

        if ($pid>0)

        {

            delete $self->{children}->{$pid};

        }

        else

        {

            last;

        }

    }

}

1;