[原创] POE学习笔记

use warnings;

use strict;



use POE;

use POE::Wheel::SocketFactory;

use POE::Wheel::ReadWrite;



POE::Session->create

	( inline_states => 

		{ _start => \&start,

		  connected => \&connected, 

		  flushed => \&flushed,

		}

	);



POE::Kernel->run;	

	

sub start {

	print "_start\n";

	my $wheel = POE::Wheel::SocketFactory->new

		( RemoteAddress => 'localhost',

		  RemotePort => 8000,

		  SuccessEvent => "connected",

		  FailureEvent => "_stop" ,

		);

	$_[HEAP]->{wheel} = $wheel;

}



sub connected {

	print "connected\n";

	my ($kernel, $heap, $socket) = @_[KERNEL, HEAP, ARG0];

	my $wheel = POE::Wheel::ReadWrite->new

		( Handle => $socket,

     	  FlushedEvent => 'flushed',

		);

	$heap->{wheel} = $wheel;

	$wheel->put("hello server");

}



sub flushed {

	print "flushed\n";

	delete $_[HEAP]->{wheel};

}

use POE;



POE::Session->create

	( inline_states => 

		{ _start => sub { print "hello world\n"; }

		}

	);

	

POE::Kernel->run;

use warnings;

use strict;



use POE;

use POE::Wheel::Somewheel;



POE::Session->create

	( inline_states => 

		{ _start => sub { 

			# do something initial

			# new some Wheel

			# put the wheel in HEAP

		  },

		  state_1 => sub {

		  	# handle state 1

		  	# new some other Wheel

		  },

		  state_2 => sub {

		    # handle state 2

		  },

		}

	);



POE::Kernel->run;

D:\MoChou\poetut>cat tpoe.pl

#!/usr/bin/perl



use warnings;

use strict;

use POE;



POE::Session->create (

    inline_states =>

    {

        _start => \&session_start,

        _stop  => \&session_stop,

        count  => \&session_count,

    }

);



print "启动 POE 内核...\n";

POE::Kernel->run();

print "POE 内核运行结束。\n";

exit;



sub session_start {

    print "Session 启动。Session ID = ", $_[SESSION]->ID, "\n";

    $_[HEAP]->{count} = 0;

    $_[KERNEL]->yield("count");

}



sub session_stop {

    print "Session 停止。Session ID = ", $_[SESSION]->ID, ".\n";

}



sub session_count {

    my ( $kernel, $heap ) = @_[ KERNEL, HEAP ];

    my $session_id = $_[SESSION]->ID;



    my $count = ++$heap->{count};

    print "数数 $count\n";



    $kernel->yield("count") if $count < 10;

}



D:\MoChou\poetut>tpoe

Session 启动。Session ID = 2

启动 POE 内核...

数数 1

数数 2

数数 3

数数 4

数数 5

数数 6

数数 7

数数 8

数数 9

数数 10

Session 停止。Session ID = 2.

POE 内核运行结束。



D:\MoChou\poetut>

# 加载 POE 以及本程序中所使用到的组件

# 之所以要写这么多是因为有些动态加载的模块 PerlAPP 识别不到，

# 编译后缺少东西不能正常运行，所以必须得写明了。

use POE;

use POE::Session;

use POE::Loop::Select;

use POE::Wheel::SocketFactory;

use POE::Wheel::ReadWrite;

use POE::Driver::SysRW;

use POE::Filter::Line;

use POE::Filter::Stream;

use POE::Resource::Aliases;

use POE::Resource::Events;

use POE::Resource::Extrefs;

use POE::Resource::FileHandles;

use POE::Resource::SIDs;

use POE::Resource::Sessions;

use POE::Resource::Signals;

use POE::Resource::Statistics;

use POE::Resource::Controls;

use POE::Wheel;

use warnings;

use strict;



use IO::Socket;

use POE qw /Wheel::ListenAccept Wheel::ReadWrite/;



# 创建监听Socket及处理Session

POE::Session->create

	( inline_states =>

		{ _start => \&start_server,

		  new_connected => \&new_connected,

		  client_input => \&client_input,

		}

	);



POE::Kernel->run;	



sub start_server {

	my ($kernel, $heap) = @_[KERNEL, HEAP];

	my $server = IO::Socket::INET->new

		( LocalPort => 8000,

          Listen => 16,

          Reuse  => "yes",

        ) or die "can't make server socket: $@\n";

        

	$heap->{server} = POE::Wheel::ListenAccept->new

		( Handle => $server, 

		  AcceptEvent => 'new_connected',

		);

}



sub new_connected {

	my ($heap, $client) = @_[HEAP, ARG0];

	my $wheel = POE::Wheel::ReadWrite->new

		( Handle => $client,

		  InputEvent => 'client_input',

		);

	# 系统中每个wheel的ID是唯一的

	$heap->{client}->{ $wheel->ID } = $wheel;

}



sub client_input {

	my ($heap, $input, $wid) = @_[HEAP, ARG0, ARG1];

	# 广播数据。如果愿意，可以屏蔽掉$wid，即发送消息的客户端

	map { $heap->{client}->{$_}->put( $input ) } keys %{$heap->{client}};

}

use warnings;

use strict;



use IO::Socket;

use POE qw /Wheel::SocketFactory Wheel::ReadWrite Wheel::ReadLine/;



POE::Session->create

	( inline_states => 

		{ _start => \&start_chat,

		  connected => \&connected,

		  connect_fail => \&connect_fail,

		  server_input => \&server_input,

		  user_input => \&user_input,

		}

	);

	

POE::Kernel->run;



sub start_chat {

	my $wheel = POE::Wheel::SocketFactory->new

		( RemoteAddress => 'localhost',

		  RemotePort => 8000,

		  SuccessEvent => "connected",

		  FailureEvent => "connect_fail",

		);

	$_[HEAP]->{server} = $wheel;

}	



sub connected {

	my ($kernel, $heap, $socket) = @_[KERNEL, HEAP, ARG0];

	my $wheel = POE::Wheel::ReadWrite->new

		( Handle => $socket,

		  InputEvent => "server_input",

		  ErrorEvent => "error_happened",

		);

	$heap->{server} = $wheel;

	

	my $console = POE::Wheel::ReadLine->new

		( InputEvent => 'user_input'

		);

	# 告诉ReadLine监控终端

	$console->get( 'input your message, bye to quit: ');

	$heap->{console} = $console;

}



sub connect_fail {

	delete $_[HEAP]->{server};

}



sub server_input {

	my ($heap, $input) = @_[HEAP, ARG0];

	# 如果使用print "$input\n"会搞乱终端

	$heap->{console}->put( $input );

}



sub user_input {

	my ($heap, $input) = @_[HEAP, ARG0];

	if ($input =~ /(quit)|(exit)|(bye)/i) {

		delete $heap->{server};

		delete $heap->{console};

		return;

	}

	# 发送到服务端

	$heap->{server}->put( $input );

	# 继续监控终端

	$heap->{console}->get( 'input your message, bye to quit: ');

}

use warnings;

use strict;



use IO::Socket;

use POE qw /Wheel::SocketFactory Wheel::ReadWrite Wheel::ReadLine/;



my $name = shift;

$name = "Anonymous" unless $name;



# 设计自己Filter

@WhoSaidFilter::ISA = qw (POE::Filter);



sub WhoSaidFilter::new {

	my $type = shift;

	my %params = @_;

	my $name = delete $params{Name};

	return bless { name => $name }, $type;

}



# 重载put函数

sub WhoSaidFilter::put {

  my ($self, $bufs) = @_;



  my @raw;

  foreach (@$bufs) {

    push @raw, "$self->{name} Said: $_, IMHO\n";

  }



  \@raw;

}



POE::Session->create

	( inline_states => 

		{ _start => \&start_chat,

		  connected => \&connected,

		  connect_fail => \&connect_fail,

		  server_input => \&server_input,

		  user_input => \&user_input,

		}

	);

	

POE::Kernel->run;



sub start_chat {

	my $wheel = POE::Wheel::SocketFactory->new

		( RemoteAddress => 'localhost',

		  RemotePort => 8000,

		  SuccessEvent => "connected",

		  FailureEvent => "connect_fail",

		);

	$_[HEAP]->{server} = $wheel;

}	



sub connected {

	my ($kernel, $heap, $socket) = @_[KERNEL, HEAP, ARG0];

	my $wheel = POE::Wheel::ReadWrite->new

		( Handle => $socket,

		  InputEvent => "server_input",

		  ErrorEvent => "error_happened",

		  # 使用输出Filter

		  OutputFilter => WhoSaidFilter->new( Name => $name ),

		);

	# 下面这一句同时把以前的那个SocketFactory的轮子删掉

	$heap->{server} = $wheel;

	

	my $console = POE::Wheel::ReadLine->new

		( InputEvent => 'user_input'

		);

	# 告诉ReadLine监控终端

	$console->get( 'input your message, bye to quit: ');

	$heap->{console} = $console;

}



sub connect_fail {

	delete $_[HEAP]->{server};

}



sub server_input {

	my ($heap, $input) = @_[HEAP, ARG0];

	# 如果使用print "$input\n"会搞乱终端

	$heap->{console}->put( $input );

}



sub user_input {

	my ($heap, $input) = @_[HEAP, ARG0];

	if ($input =~ /(quit)|(exit)|(bye)/i) {

		delete $heap->{server};

		delete $heap->{console};

		return;

	}

	# 发送到服务端

	$heap->{server}->put( $input );

	# 继续监控终端

	$heap->{console}->get( 'input your message, bye to quit: ');

}

use warnings;

use strict;



package Component::TimerRoutine;



use Carp qw /croak/;

use POE;



# 启动一个Session，执行定时任务

sub spawn {

	my $type = shift;

	my %params = @_;

	my $interval = delete $params{Interval};

	my $workhorse = delete $params{Workhorse};

	croak "Workhorse param must be a subroutine reference" 

		unless ref($workhorse) eq 'CODE';

	POE::Session->create

		( inline_states => {

			_start => \&timer_routine_start,

			dowork => \&dowork,

		  },

		  # 在创建一个Session时，可以通过heap参数设定Session私有数据

		  heap => {

		  	interval => $interval,

		  	workhorse => $workhorse,

		  }

			

		);

	undef;

}



sub timer_routine_start {

	my ($kernel, $heap) = @_[KERNEL, HEAP];

	if ($heap->{interval}) {

		$kernel->delay_set( 'dowork', $heap->{interval} );

	} else {

		$kernel->yield( 'dowork' );

	}

}



sub dowork {

	my ($kernel, $heap) = @_[KERNEL, HEAP];

	return unless $heap->{workhorse}->();

	if ($heap->{interval}) {

		$kernel->delay_set( 'dowork', $heap->{interval} );

	} else {

		$kernel->yield( 'dowork' );

	}

}





package main;



use POE;



sub work {

	print time, "\n";

	return 1;

}



Component::TimerRoutine->spawn

	( Interval => 1,

	  Workhorse => \&work,

	);



POE::Kernel->run;

use warnings;

use strict;



use IO::Socket;

use POE qw /Component::Server::TCP/;



# 客户连接表

my %conntab = ();



POE::Component::Server::TCP->new

	( Port => 8000,

	  ClientConnected => sub {

	  	# 登记客户

	  	$conntab{ $_[HEAP]->{client}->ID } = $_[HEAP]->{client}; 

	  },

	  ClientDisconnected => sub {

	  	# 注销客户

	  	delete $conntab{ $_[HEAP]->{client}->ID }; 

	  },

	  ClientInput => \&client_input,

	);



POE::Kernel->run;



sub client_input {

	my ($heap, $input, $wid) = @_[HEAP, ARG0, ARG1];

	# 广播数据。如果愿意，可以屏蔽掉$wid，即发送消息的客户端

	map { $conntab{$_}->put( $input ) } keys %conntab;

}



# 是的，这些代码实现了一个聊天室服务器

#!/usr/bin/perl



use warnings;

use strict;



use POE qw(Component::Server::TCP);



POE::Component::Server::TCP->new(

    Port => 12345,

    ClientInput => \&client_input,

);



POE::Kernel->run();

exit;



sub client_input {

    my ( $heap, $input ) = @_[ HEAP, ARG0 ];

    $heap->{client}->put($input);

}