Mercurial > hg > nginx-tests

comparison stream_upstream.t @ 816:77359b849cd5

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Tests: stream package.
author Andrey Zelenkov <zelenkov@nginx.com>
date Mon, 20 Jul 2015 15:06:09 +0300
parents 824754da4afc
children e9064d691790
comparison
equal deleted inserted replaced
815:9f5f604a840e 816:77359b849cd5
16 16
17 BEGIN { use FindBin; chdir($FindBin::Bin); } 17 BEGIN { use FindBin; chdir($FindBin::Bin); }
18 18
19 use lib 'lib'; 19 use lib 'lib';
20 use Test::Nginx; 20 use Test::Nginx;
21 use Test::Nginx::Stream qw/ stream /;
21 22
22 ############################################################################### 23 ###############################################################################
23 24
24 select STDERR; $| = 1; 25 select STDERR; $| = 1;
25 select STDOUT; $| = 1; 26 select STDOUT; $| = 1;
103 my (%ports, $peer); 104 my (%ports, $peer);
104 105
105 $peer = $opts{peer}; 106 $peer = $opts{peer};
106 107
107 for (1 .. $count) { 108 for (1 .. $count) {
108 if (stream_get($data, $peer) =~ /(\d+)/) { 109 if (stream($peer)->io($data) =~ /(\d+)/) {
109 $ports{$1} = 0 unless defined $ports{$1}; 110 $ports{$1} = 0 unless defined $ports{$1};
110 $ports{$1}++; 111 $ports{$1}++;
111 } 112 }
112 } 113 }
113 114
114 return join ', ', map { $_ . ": " . $ports{$_} } sort keys %ports; 115 return join ', ', map { $_ . ": " . $ports{$_} } sort keys %ports;
115 }
116
117 sub stream_get {
118 my ($data, $peer) = @_;
119
120 my $s = stream_connect($peer);
121 stream_write($s, $data);
122
123 $data = '';
124 while (my $buf = stream_read($s)) {
125 $data .= $buf;
126 }
127 return $data;
128 }
129
130 sub stream_connect {
131 my $peer = shift;
132 my $s = IO::Socket::INET->new(
133 Proto => 'tcp',
134 PeerAddr => $peer
135 )
136 or die "Can't connect to nginx: $!\n";
137
138 return $s;
139 }
140
141 sub stream_write {
142 my ($s, $message) = @_;
143
144 local $SIG{PIPE} = 'IGNORE';
145
146 $s->blocking(0);
147 while (IO::Select->new($s)->can_write(1.5)) {
148 my $n = $s->syswrite($message);
149 last unless $n;
150 $message = substr($message, $n);
151 last unless length $message;
152 }
153
154 if (length $message) {
155 $s->close();
156 }
157 }
158
159 sub stream_read {
160 my ($s) = @_;
161 my ($buf);
162
163 $s->blocking(0);
164 if (IO::Select->new($s)->can_read(5)) {
165 $s->sysread($buf, 1024);
166 };
167
168 log_in($buf);
169 return $buf;
170 } 116 }
171 117
172 ############################################################################### 118 ###############################################################################
173 119
174 sub stream_daemon { 120 sub stream_daemon {