1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
|
// This code has to work both in 'roxen.pike' and all modules
// string _cvs_version = "$Id: socket.pike,v 1.12 1998/03/26 07:51:45 per Exp $";
#if !efun(roxen)
#define roxen roxenp()
#endif
#if DEBUG_LEVEL > 19
#ifndef SOCKET_DEBUG
# define SOCKET_DEBUG
#endif
#endif
private void connected(array args)
{
if (!args) {
#ifdef SOCKET_DEBUG
perror("SOCKETS: async_connect: No arguments to connected\n");
#endif /* SOCKET_DEBUG */
return;
}
#ifdef SOCKET_DEBUG
perror("SOCKETS: async_connect ok.\n");
#endif
args[2]->set_id(0);
args[0](args[2], @args[1]);
}
private void failed(array args)
{
#ifdef SOCKET_DEBUG
perror("SOCKETS: async_connect failed\n");
#endif
args[2]->set_id(0);
destruct(args[2]);
args[0](0, @args[1]);
}
private void got_host_name(string host, string oh, int port,
function callback, mixed ... args)
{
object f;
f=Stdio.File();
#ifdef SOCKET_DEBUG
perror("SOCKETS: async_connect "+oh+" == "+host+"\n");
#endif
if(!f->open_socket())
{
#ifdef SOCKET_DEBUG
perror("SOCKETS: socket() failed. Out of sockets?\n");
#endif
callback(0, @args);
destruct(f);
return;
}
f->set_id( ({ callback, args, f }) );
f->set_nonblocking(0, connected, failed);
#ifdef FD_DEBUG
mark_fd(f->query_fd(), "async socket communication: -> "+host+":"+port);
#endif
if(catch(f->connect(host, port))) // Illegal format...
{
#ifdef SOCKET_DEBUG
perror("SOCKETS: Illegal internet address in connect in async comm.\n");
#endif
callback(0, @args);
destruct(f);
return;
}
}
void async_connect(string host, int port, function|void callback,
mixed ... args)
{
#ifdef SOCKET_DEBUG
perror("SOCKETS: async_connect requested to "+host+":"+port+"\n");
#endif
roxen->host_to_ip(host, got_host_name, host, port, callback, @args);
}
private void my_pipe_done(object which)
{
if(objectp(which))
{
if(which->done_callback)
which->done_callback(which);
else
destruct(which);
}
}
void async_pipe(object to, object from, function|void callback,
mixed|void id, mixed|void cl, mixed|void file)
{
object pipe=Pipe.pipe();
object cache;
#ifdef SOCKET_DEBUG
perror("async_pipe(): ");
#endif
if(callback)
pipe->set_done_callback(callback, id);
else if(cl) {
cache = roxen->cache_file(cl, file);
if(cache)
{
#ifdef SOCKET_DEBUG
perror("Using normal pipe with done callback.\n");
#endif
pipe->input(cache->file);
pipe->set_done_callback(my_pipe_done, cache);
pipe->output(to);
destruct(from);
pipe->start();
return;
}
if(cache = roxen->create_cache_file(cl, file))
{
#ifdef SOCKET_DEBUG
perror("Using normal pipe with cache.\n");
#endif
pipe->output(cache->file);
pipe->set_done_callback(my_pipe_done, cache);
pipe->input(from);
pipe->output(to);
return;
}
}
#ifdef SOCKET_DEBUG
perror("Using normal pipe.\n");
#endif
pipe->input(from);
pipe->output(to);
}
void async_cache_connect(string host, int port, string cl,
string entry, function|void callback,
mixed ... args)
{
object cache;
#ifdef SOCKET_DEBUG
perror("SOCKETS: async_cache_connect requested to "+host+":"+port+"\n");
#endif
cache = roxen->cache_file(cl, entry);
if(cache)
{
object f;
f=cache->file;
// perror("Cache file is %O\n", f);
cache->file = 0; // do _not_ close the actual file when returning...
destruct(cache);
return callback(f, @args);
}
roxen->host_to_ip(host, got_host_name, host, port, callback, @args);
}
|
