# Copyright (c) 2003-2009 Nokia Corporation and/or its subsidiary(-ies).

# All rights reserved.

# This component and the accompanying materials are made available

# under the terms of "Eclipse Public License v1.0"

# which accompanies this distribution, and is available

# at the URL "http://www.eclipse.org/legal/epl-v10.html".

#

# Initial Contributors:

# Nokia Corporation - initial contribution.

#

# Contributors:

#

# Description:

#

package Msg;

use strict;

use IO::Select;

use IO::Socket;

use Carp;

use vars qw ( %scan_retrieves %publish_retrieves $scan_manages $publish_manages);

 %scan_retrieves = ();

%publish_retrieves = ();

$scan_manages   = IO::Select->new();

$publish_manages   = IO::Select->new();

my $obstructing_maintained = 0;

my $AllAssociations = 0;

BEGIN {

    # Checks if blocking is supported

    eval {

        require POSIX; POSIX->import(qw (F_SETFL O_NONBLOCK EAGAIN));

    };

    $obstructing_maintained = 1 unless $@;

}

use Socket qw(SO_KEEPALIVE SOL_SOCKET);

use constant TCP_KEEPIDLE  => 4; # Start keeplives after this period

use constant TCP_KEEPINTVL => 5; # Interval between keepalives

use constant TCP_KEEPCNT   => 6; # Number of keepalives before death

# AllAssociations

#

# Inputs

#

# Outputs

#

# Description

# This function returns the total number of connections

sub AllAssociations

{

  return $AllAssociations;

}

# associate

#

# Inputs

# $collection

# $toReceiver (Host associate to)

# $toChange (Port number to associate to)

# $get_notice_process (Function to call on recieving data)

#

# Outputs

#

# Description

# This function connects the client to the server

sub associate {

    my ($collection, $toChange, $toReceiver, $get_notice_process) = @_;

    # Create a new internet socket

    my $link = IO::Socket::INET->new (

                                      PeerAddr => $toReceiver,

                                      PeerPort => $toChange,

                                      Proto    => 'tcp',

                                      TimeOut => 10,

                                      Reuse    => 1);

    return undef unless $link;

    # Set KeepAlive

    setsockopt($link, SOL_SOCKET, SO_KEEPALIVE,  pack("l", 1));

    setsockopt($link, &Socket::IPPROTO_TCP, TCP_KEEPIDLE,  pack("l", 30));

    setsockopt($link, &Socket::IPPROTO_TCP, TCP_KEEPCNT,   pack("l", 2));

    setsockopt($link, &Socket::IPPROTO_TCP, TCP_KEEPINTVL, pack("l", 30));

    # Increse the total connection count

    $AllAssociations++;

    # Create a connection end-point object

    my $asso = bless {

        sock                   => $link,

        rcvd_notification_proc => $get_notice_process,

    }, $collection;

      # Set up the callback to the rcv function

    if ($get_notice_process) {

        my $retrieve = sub {_get($asso, 0)};

        define_result_manager ($link, "read" => $retrieve);

    }

    $asso;

}

# unplug

#

# Inputs

# $asso (Connection object)

#

# Outputs

#

# Description

# This function disconnects a connection and cleans up

sub unplug {

    my $asso = shift;

    # Decrease the number of total connections

    $AllAssociations--;

    # Delete the socket

    my $link = delete $asso->{sock};

    return unless defined($link);

    # Set to not try and check for reads and writes of this socket

    define_result_manager ($link, "write" => undef, "read" => undef);

    close($link);

}

# transmit_immediately

#

# Inputs

# $asso (Connection object)

# $content (Message to send)

#

# Outputs

#

# Description

# This function does a immediate send, this will block if the socket is not writeable

sub transmit_immediately {

    my ($asso, $content) = @_;

    # Puts the message in the queue

    _lineup ($asso, $content);

    # Flushes the queue

    $asso->_transmit (1); # 1 ==> flush

}

# transmit_afterwards

#

# Inputs

# $asso (Connection object)

# $content (Message to send)

#

# Outputs

#

# Description

# This function does a sends at a later time, does not block if the socket is not writeable.

# It sets a callback to send the data in the queue when the socket is writeable

sub transmit_afterwards {

    my ($asso, $content) = @_;

    # Puts the message in the queue

    _lineup($asso, $content);

    # Get the current socket

    my $link = $asso->{sock};

    return unless defined($link);

    # Sets the callback to send the data when the socket is writeable

    define_result_manager ($link, "write" => sub {$asso->_transmit(0)});

}

# _lineup

#

# Inputs

# $asso (Connection object)

# $content (Message to send)

#

# Outputs

#

# Description

# This is a private function to place the message on the queue for this socket

sub _lineup {

    my ($asso, $content) = @_;

    # prepend length (encoded as network long)

    my $dist = length($content);

    # Stores the length as a network long in the first 4 bytes of the message

    $content = pack ('N', $dist) . $content; 

    push (@{$asso->{queue}}, $content);

}

# _transmit

#

# Inputs

# $asso (Connection object)

# $remove (Deferred Mode)

#

# Outputs

#

# Description

# This is a private function sends the data

sub _transmit {

    my ($asso, $remove) = @_;

    my $link = $asso->{sock};

    return unless defined($link);

    my ($Lrq) = $asso->{queue};

    # If $remove is set, set the socket to blocking, and send all

    # messages in the queue - return only if there's an error

    # If $remove is 0 (deferred mode) make the socket non-blocking, and

    # return to the event loop only after every message, or if it

    # is likely to block in the middle of a message.

    $remove ? $asso->define_obstructing() : $asso->define_not_obstructing();

    my $branch = (exists $asso->{send_offset}) ? $asso->{send_offset} : 0;

    # Loop through the messages in the queue

    while (@$Lrq) {

        my $content            = $Lrq->[0];

        my $sequencetoPublish = length($content) - $branch;

        my $sequence_published  = 0;

        while ($sequencetoPublish) {

            $sequence_published = syswrite ($link, $content,

                                       $sequencetoPublish, $branch);

            if (!defined($sequence_published)) {

                if (_faultwillObstruct($!)) {

                    # Should happen only in deferred mode. Record how

                    # much we have already sent.

                    $asso->{send_offset} = $branch;

                    # Event handler should already be set, so we will

                    # be called back eventually, and will resume sending

                    return 1;

                } else {    # Uh, oh

                    $asso->manage_transmitted_fault($!);

                    return 0; # fail. Message remains in queue ..

                }

            }

            $branch         += $sequence_published;

            $sequencetoPublish -= $sequence_published;

        }

        delete $asso->{send_offset};

        $branch = 0;

        shift @$Lrq;

        last unless $remove; # Go back to select and wait

                            # for it to fire again.

    }

    # Call me back if queue has not been drained.

    if (@$Lrq) {

        define_result_manager ($link, "write" => sub {$asso->_transmit(0)});

    } else {

        define_result_manager ($link, "write" => undef);

    }

    1;  # Success

}

# _faultwillObstruct

#

# Inputs

# $asso (Connection object)

#

# Outputs

#

# Description

# This is a private function processes the blocking error message

sub _faultwillObstruct {

    if ($obstructing_maintained) {

        return ($_[0] == EAGAIN());

    }

    return 0;

}

# define_not_obstructing

#

# Inputs

# $_[0] (Connection socket)

#

# Outputs

#

# Description

# This is a function set non-blocking on a socket

sub define_not_obstructing {                        # $asso->define_obstructing

    if ($obstructing_maintained) {

        # preserve other fcntl flags

        my $pins = fcntl ($_[0], F_GETFL(), 0);

        fcntl ($_[0], F_SETFL(), $pins | O_NONBLOCK());

    }

}

# define_obstructing

#

# Inputs

# $_[0] (Connection socket)

#

# Outputs

#

# Description

# This is a function set blocking on a socket

sub define_obstructing {

    if ($obstructing_maintained) {

        my $pins = fcntl ($_[0], F_GETFL(), 0);

        $pins  &= ~O_NONBLOCK(); # Clear blocking, but preserve other flags

        fcntl ($_[0], F_SETFL(), $pins);

    }

}

# manage_transmitted_fault

#

# Inputs

# $asso (Connection object)

# $fault_content (Error message)

#

# Outputs

#

# Description

# This is a function warns on send errors and removes the socket from list of writable sockets

sub manage_transmitted_fault {

   # For more meaningful handling of send errors, subclass Msg and

   # rebless $asso.  

   my ($asso, $fault_content) = @_;

   warn "Error while sending: $fault_content \n";

   define_result_manager ($asso->{sock}, "write" => undef);

}

#-----------------------------------------------------------------

# Receive side routines

# recent_agent

#

# Inputs

# $collection (Package)

# $mi_receiver (Hostname of the interface to use)

# $mi_change (Port number to listen on)

# $enter_process (Reference to function to call when accepting a connection)

#

# Outputs

#

# Description

# This is a function create a listening socket

my ($g_enter_process,$g_collection);

my $primary_plug = 0;

sub recent_agent {

    @_ >= 4 || die "Msg->recent_agent (myhost, myport, login_proc)\n";

    my ($RepeatNumber);

    my ($collection, $changes, $mi_receiver, $enter_process, $iAssociationBreak, $PlugAssociations) = @_;

    # Set a default Socket timeout value

    $iAssociationBreak = 0 if (!defined $iAssociationBreak);

    # Set a default Socket retry to be forever

    $PlugAssociations = -1 if (!defined $PlugAssociations);

    while(!$primary_plug)

    {

        #Check to see if there is a retry limit and if the limit has been reached

        if ($PlugAssociations != -1)

        {

            if (($RepeatNumber / scalar(@$changes)) >= $PlugAssociations)

            {

                die "ERROR: could not create socket after ".$RepeatNumber / scalar(@$changes)." attempts";            

            } else {

                # Increment the number of retries

                $RepeatNumber++;

            }

        }

        #Try the first port on the list

        my $mi_change = shift(@$changes);

        #Place the port on the back of the queue

        push @$changes,$mi_change;

        print "Using port number $mi_change\n";

        $primary_plug = IO::Socket::INET->new (

                                              LocalAddr => $mi_receiver,

                                              LocalPort => $mi_change,

                                              Listen    => 5,

                                              Proto     => 'tcp',

                                              TimeOut =>    10,

                                              Reuse     => 1);

        sleep $iAssociationBreak if (!$primary_plug);

    }

    # Set KeepAlive

    setsockopt($primary_plug, SOL_SOCKET, SO_KEEPALIVE,  pack("l", 1));

    setsockopt($primary_plug, &Socket::IPPROTO_TCP, TCP_KEEPIDLE,  pack("l", 30));

    setsockopt($primary_plug, &Socket::IPPROTO_TCP, TCP_KEEPCNT,   pack("l", 2));

    setsockopt($primary_plug, &Socket::IPPROTO_TCP, TCP_KEEPINTVL, pack("l", 30));

    # Add the socket to the list on filehandles to read from.

    define_result_manager ($primary_plug, "read" => \&_recent_node);

    # Store the package name and login proc for later use

    $g_enter_process = $enter_process; $g_collection = $collection;

}

sub get_immediately {

    my ($asso) = @_;

    my ($content, $fault) = _get ($asso, 1); # 1 ==> rcv now

    return wantarray ? ($content, $fault) : $content;

}

sub _get {                     # Complement to _transmit

    my ($asso, $get_immediately) = @_; # $get_immediately complement of $remove

    # Find out how much has already been received, if at all

    my ($content, $branch, $sequencetoScan, $sequence_scan);

    my $link = $asso->{sock};

    return unless defined($link);

    if (exists $asso->{msg}) {

        $content           = $asso->{msg};

        $branch        = length($content) - 1;  # sysread appends to it.

        $sequencetoScan = $asso->{bytes_to_read};

        delete $asso->{'msg'};              # have made a copy

    } else {

        # The typical case ...

        $content           = "";                # Otherwise -w complains 

        $branch        = 0 ;  

        $sequencetoScan = 0 ;                # Will get set soon

    }

    # We want to read the message length in blocking mode. Quite

    # unlikely that we'll get blocked too long reading 4 bytes

    if (!$sequencetoScan)  {                 # Get new length 

        my $storage;

        $asso->define_obstructing();

        $sequence_scan = sysread($link, $storage, 4);

        if ($! || ($sequence_scan != 4)) {

            goto FINISH;

        }

        $sequencetoScan = unpack ('N', $storage);

    }

    $asso->define_not_obstructing() unless $get_immediately;

    while ($sequencetoScan) {

        $sequence_scan = sysread ($link, $content, $sequencetoScan, $branch);

        if (defined ($sequence_scan)) {

            if ($sequence_scan == 0) {

                last;

            }

            $sequencetoScan -= $sequence_scan;

            $branch        += $sequence_scan;

        } else {

            if (_faultwillObstruct($!)) {

                # Should come here only in non-blocking mode

                $asso->{msg}           = $content;

                $asso->{bytes_to_read} = $sequencetoScan;

                return ;   # .. _get will be called later

                           # when socket is readable again

            } else {

                last;

            }

        }

    }

  FINISH:

    if (length($content) == 0) {

        $asso->unplug();

    }

    if ($get_immediately) {

        return ($content, $!);

    } else {

        &{$asso->{rcvd_notification_proc}}($asso, $content, $!);

    }

}

sub _recent_node {

    my $link = $primary_plug->accept();

    $AllAssociations++;

    my $asso = bless {

        'sock' =>  $link,

        'state' => 'connected'

    }, $g_collection;

    my $get_notice_process =

        &$g_enter_process ($asso, $link->peerhost(), $link->peerport());

    if ($get_notice_process) {

        $asso->{rcvd_notification_proc} = $get_notice_process;

        my $retrieve = sub {_get($asso,0)};

        define_result_manager ($link, "read" => $retrieve);

    } else {  # Login failed

        $asso->unplug();

    }

}

#----------------------------------------------------

# Event loop routines used by both client and server

sub define_result_manager {

    shift unless ref($_[0]); # shift if first arg is package name

    my ($manage, %parameters) = @_;

    my $retrieve;

    if (exists $parameters{'write'}) {

        $retrieve = $parameters{'write'};

        if ($retrieve) {

            $publish_retrieves{$manage} = $retrieve;

            $publish_manages->add($manage);

        } else {

            delete $publish_retrieves{$manage};

            $publish_manages->remove($manage);

        }

    }

    if (exists $parameters{'read'}) {

        $retrieve = $parameters{'read'};

        if ($retrieve) {

            $scan_retrieves{$manage} = $retrieve;

            $scan_manages->add($manage);

        } else {

            delete $scan_retrieves{$manage};

            $scan_manages->remove($manage);

       }

    }

}

sub result_iteration {

    my ($collection, $starting_scan_break, $iteration_number) = @_; # result_iteration(1) to process events once

    my ($asso, $scan, $publish, $scandefine, $publishdefine);

    while (1) {

        # Quit the loop if no handles left to process

        last unless ($scan_manages->count() || $publish_manages->count());

        if (defined $starting_scan_break)

        {

            ($scandefine, $publishdefine) = IO::Select->select ($scan_manages, $publish_manages, undef, $starting_scan_break);

            # On initial timeout a read expect a read within timeout if not disconnect

            if (!defined $scandefine)

            {

              print "WARNING: no response from server within $starting_scan_break seconds\n";

              last;

            }

            # Unset intial timeout

            $starting_scan_break = undef;

        } else {

            ($scandefine, $publishdefine) = IO::Select->select ($scan_manages, $publish_manages, undef, undef);

        }

        foreach $scan (@$scandefine) {

            &{$scan_retrieves{$scan}} ($scan) if exists $scan_retrieves{$scan};

        }