Skip to content

Commit d5dd5b3

Browse files
EDBCREPOEDBCREPO
committed
event.h | rewritten code
1 parent eb33f4e commit d5dd5b3

6 files changed

Lines changed: 137 additions & 81 deletions

File tree

‎include/nodepp/crypto.h‎

Lines changed: 8 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -83,8 +83,7 @@ namespace nodepp { class hash_t {
8383
}; ptr_t<NODE> obj;
8484

8585
string_t hex() const noexcept {
86-
free(); return { (char*) &obj->bff, obj->length };
87-
}
86+
free(); return string_t( (char*) &obj->bff, obj->length ); }
8887

8988
public:
9089

@@ -140,8 +139,7 @@ namespace nodepp { class hmac_t {
140139
}; ptr_t<NODE> obj;
141140

142141
string_t hex() const noexcept {
143-
free(); return { (char*) &obj->bff, obj->length };
144-
}
142+
free(); return string_t( (char*) &obj->bff, obj->length ); }
145143

146144
public:
147145

@@ -176,8 +174,7 @@ namespace nodepp { class hmac_t {
176174
}
177175

178176
string_t get() const noexcept {
179-
return encoder::base16::atob( this->hex() );
180-
}
177+
return encoder::base16::atob( this->hex() ); }
181178

182179
bool is_available() const noexcept { return obj->state == 1; }
183180

@@ -228,19 +225,19 @@ namespace nodepp { class xor_t {
228225

229226
void update( string_t msg ) const noexcept {
230227
if( !obj->state ){ return; } ulong chunk = NODEPP_CHUNK_SIZE;
231-
while( !msg.empty() ){ auto tmp = msg.slice( 0, chunk );
232-
forEach( x, tmp ){ CTX &y = obj->ctx[0];
233-
x ^= y.key[ y.pos % y.key.size() ]; ++y.pos;
228+
while( !msg.empty() ) { auto tmp = msg.slice( 0, chunk );
229+
for ( auto &x: tmp ) { CTX &y = obj->ctx [ 0 ];
230+
x ^= y.key[ y.pos ]; y.pos += ( y.pos+1 )% y.key.size();
234231
} if ( tmp .empty() ) { return; }
235232
elif ( onData.empty() ) { obj->bff +=tmp; }
236233
else { onData.emit(tmp); } msg.ptr().slice( chunk, (ulong)-1 ); }
237234
}
238235

239236
bool is_available() const noexcept { return obj->state == 1; }
240237

241-
bool is_closed() const noexcept { return obj->state == 0; }
238+
bool is_closed() const noexcept { return obj->state == 0; }
242239

243-
string_t get() const noexcept { free(); return obj->bff; }
240+
string_t get() const noexcept { free(); return obj->bff; }
244241

245242
void free() const noexcept {
246243
if( obj->state == 0 ){ return; }

‎include/nodepp/event.h‎

Lines changed: 37 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -17,10 +17,10 @@
1717
namespace nodepp { template< class... A > class event_t {
1818
protected:
1919

20-
using DONE = function_t<bool,A...>;
2120
struct NODE {
22-
queue_t<DONE> que; uchar state=0x00;
23-
}; ptr_t <NODE> obj;
21+
queue_t<function_t<int,A...>> que;
22+
uchar state = 0x00;
23+
}; ptr_t <NODE> obj;
2424

2525
enum STATE : uchar {
2626
EV_STATE_UNKNOWN = 0b00000000,
@@ -44,64 +44,60 @@ namespace nodepp { template< class... A > class event_t {
4444
ptr_t<task_t> add ( function_t<int,A...> cb ) const noexcept {
4545
ptr_t<task_t> task( 0UL, task_t() );
4646

47-
obj->que.push([=]( A... args ){ int c=-1; do {
48-
if( task.null() || cb.null() ) { return false; }
49-
if( task->flag & TASK_STATE::CLOSED ){ return false; }
50-
if( task->flag & TASK_STATE::USED ){ return true ; }
51-
task->flag|= TASK_STATE::USED ; c=cb(args...) ;
52-
if( cb.null() || task.null() ) { return false; }
53-
task->flag&=~TASK_STATE::USED;
54-
} while(c==0); return c==-1 ? false : true; });
47+
if( is_used() ){ return nullptr; }
48+
49+
obj->que.unshift([=]( A... args ){
50+
if( task->flag & TASK_STATE::CLOSED ){ return -1; }
51+
return cb(args...);
52+
});
5553

5654
task->flag = TASK_STATE::OPEN;
57-
task->addr = obj->que.last();
55+
task->addr = obj->que.last() ;
5856
task->sign = &obj;
5957

6058
return task; }
6159

6260
ptr_t<task_t> once( function_t<void,A...> cb ) const noexcept {
6361
ptr_t<task_t> task( 0UL, task_t() );
6462

65-
obj->que.push([=]( A... args ){
66-
if( task.null() || cb.null() ) /*-*/ { return false; }
67-
if( task->flag & TASK_STATE::CLOSED ){ return false; }
68-
task->flag = TASK_STATE::CLOSED ; cb(args...) ;
69-
return false; });
63+
if( is_used() ){ return nullptr; }
64+
65+
obj->que.unshift([=]( A... args ){
66+
if( task->flag & TASK_STATE::CLOSED ){ return -1; }
67+
cb(args...); return -1;
68+
});
7069

7170
task->flag = TASK_STATE::OPEN;
72-
task->addr = obj->que.last();
71+
task->addr = obj->que.last() ;
7372
task->sign = &obj;
7473

7574
return task; }
7675

7776
ptr_t<task_t> on ( function_t<void,A...> cb ) const noexcept {
7877
ptr_t<task_t> task( 0UL, task_t() );
7978

80-
obj->que.push([=]( A... args ){
81-
if( task.null() ) /*--------------*/ { return false; }
82-
if( task->flag & TASK_STATE::CLOSED ){ return false; }
83-
if( task->flag & TASK_STATE::USED ){ return true ; }
84-
task->flag|= TASK_STATE::USED ; cb(args...) ;
85-
if( cb.null() || task.null() ) { return false; }
86-
task->flag&=~TASK_STATE::USED;
87-
return true; });
79+
if( is_used() ){ return nullptr; }
80+
81+
obj->que.unshift([=]( A... args ){
82+
if( task->flag & TASK_STATE::CLOSED ){ return -1; }
83+
cb(args...); return 1;
84+
});
8885

8986
task->flag = TASK_STATE::OPEN;
90-
task->addr = obj->que.last();
87+
task->addr = obj->que.last() ;
9188
task->sign = &obj;
9289

9390
return task; }
9491

9592
/*─······································································─*/
9693

9794
void off( ptr_t<task_t> address ) const noexcept {
98-
if( address.null() ) /*--------------*/ { return; }
95+
if( address.null()||empty()||is_used() ){ return; }
9996
if( address->sign != &obj ) /*-------*/ { return; }
10097
if( address->flag & TASK_STATE::CLOSED ){ return; }
101-
address->flag = TASK_STATE::CLOSED;
102-
auto node = obj->que.as( address->addr );
103-
if( node == nullptr ) /*-------------*/ { return; }
104-
obj->que.erase( node );
98+
address->flag = TASK_STATE::CLOSED ;
99+
auto node=obj->que.as(address->addr);
100+
if( node ){ obj->que.erase( node ); }
105101
}
106102

107103
/*─······································································─*/
@@ -121,22 +117,20 @@ namespace nodepp { template< class... A > class event_t {
121117
/*─······································································─*/
122118

123119
void emit( const A&... args ) const noexcept {
124-
if( empty() || is_paused() || is_used() ){ return; }
120+
if( is_paused() || is_used() || empty() ){ return; }
125121

126-
obj->state |= STATE::EV_STATE_USED;
127-
obj->que.set( obj->que.first() );
122+
obj->state|= STATE::EV_STATE_USED;
123+
obj->que.set( obj->que.last() );
128124

129125
while( obj->que.get() != nullptr ){
126+
auto x=obj->que.get(); uchar z=0x01; auto c=x->data;
130127

131-
auto x = obj->que.get();
132-
auto y = x->next;
133-
bool val = x->data.emit( args... );
134-
135-
if( empty() ) /*---*/ { break; }
136-
if( !val ){ obj->que.erase(x); }
137-
if( y==nullptr ) /**/ { break; }
128+
if( !c.empty() ){ z = c.emit(args...)>=0 ? 0x00 : 0x01; }
129+
if( !x->prev ){ z|= 0x02; }
130+
if( z & 0x01 ){ obj->que.erase(x); }
131+
if( z & 0x02 ){ break ; }
138132

139-
obj->que.next(); }
133+
obj->que.prev(); }
140134

141135
/**/obj->state &=~ STATE::EV_STATE_USED;
142136
if( obj->state & STATE::EV_STATE_KILL ){ clear(); }

‎include/nodepp/invoker.h‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@ public: invoker_t() {}
3535

3636
bool is_valid( uchar_64 address ) const noexcept { return que.is_valid( address ); }
3737
int off ( uchar_64 address ) const noexcept { return que.remove ( address ); }
38+
handler_t<NODE_CLB> get_handler() const noexcept { return que; }
3839

3940
/*─······································································─*/
4041

‎include/nodepp/posix/socket.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -431,7 +431,7 @@ class socket_t {
431431
}
432432

433433
void set_write_address( const SOCKADDR_ST& address ) const noexcept {
434-
obj->tmp_addr = address;
434+
get_write_address() = address;
435435
}
436436

437437
SOCKADDR_ST& get_read_address() const noexcept {

‎include/nodepp/windows/socket.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -424,7 +424,7 @@ class socket_t {
424424
}
425425

426426
void set_write_address( const SOCKADDR_ST& address ) const noexcept {
427-
obj->tmp_addr = address;
427+
get_write_address() = address;
428428
}
429429

430430
SOCKADDR_ST& get_read_address() const noexcept {

‎main.cpp‎

Lines changed: 89 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -1,40 +1,104 @@
1+
#define MAX_BATCH 10
2+
3+
/*────────────────────────────────────────────────────────────────────────────*/
4+
15
#include <nodepp/nodepp.h>
2-
#include <nodepp/https.h>
3-
#include <nodepp/crypto.h>
6+
#include <nodepp/worker.h>
7+
#include <nodepp/timer.h>
8+
#include <nodepp/http.h>
9+
#include <nodepp/path.h>
10+
#include <nodepp/ws.h>
11+
#include <nodepp/fs.h>
12+
13+
/*────────────────────────────────────────────────────────────────────────────*/
414

515
using namespace nodepp;
616

7-
void onMain(){
17+
/*────────────────────────────────────────────────────────────────────────────*/
18+
19+
void server(){
820

9-
ssl_t ssl; // ( "./ssl/cert.key", "./ssl/cert.crt" );
21+
auto server = http::server([=]( http_t cli ){
1022

11-
fetch_t args;
12-
args.method = "GET";
13-
args.url = "https://www.mamp.one/wp-content/uploads/2024/09/image-resources2.jpg";
14-
args.headers = header_t({
15-
{ "Host", url::host(args.url) }
16-
});
17-
18-
// args.body = "MYBODY";
23+
cli.write_header( 200, header_t({
24+
{ "Content-Security-Policy", "*" }
25+
}) );
1926

20-
https::fetch( args, &ssl )
27+
cli.write("Hello World!");
2128

22-
.then([]( https_t cli ){
29+
}); ws::server( server );
2330

24-
auto b64 = crypto::encoder::BASE64();
25-
auto out = fs::writable( "index.html" );
31+
server.onConnect([=]( ws_t cli ){
32+
33+
console::log("connected");
2634

27-
cli.onData ([=]( string_t data ){ b64.update(data); });
28-
b64.onData ([=]( string_t data ){ out.write (data); });
29-
b64.onClose([=](){ out.write( "'>" ); });
35+
cli.onData([=]( string_t data ){
36+
console::log( data );
37+
cli.close(); // << - stress
38+
});
39+
40+
cli.onClose([=](){
41+
console::log("closed");
42+
});
43+
44+
timer::add([=](){
45+
cli.write( string::format( "server %lu", process::now() ) );
46+
return cli.is_closed() ? -1 : 1; }, 10 );
47+
48+
});
49+
50+
server.onError([=]( except_t err ){
51+
console::log( ">>", err.what() );
52+
});
53+
54+
server.listen( "localhost", 8000, [=]( socket_t server ){
55+
console::log("server started at http://localhost:8000");
56+
});
57+
58+
}
3059

31-
out.write( "<img src='data:image/png;base64, " );
32-
stream::pipe( cli );
60+
/*────────────────────────────────────────────────────────────────────────────*/
3361

34-
})
62+
void client() {
3563

36-
.fail([]( except_t err ){
37-
console::error( err );
64+
auto client = ws::client( "ws://localhost:8000/" );
65+
66+
client.onConnect([=]( ws_t cli ){
67+
68+
cli.write( string::format( "client %lu", process::now() ) );
69+
cli.close();
70+
71+
});
72+
73+
client.onError([=]( except_t err ){
74+
console::log( "<>", err.data() );
3875
});
3976

40-
}
77+
}
78+
79+
/*────────────────────────────────────────────────────────────────────────────*/
80+
81+
void onMain() {
82+
83+
worker::add( coroutine::add( COROUTINE(){
84+
coBegin /*--*/ ; server();
85+
process::wait(); coFinish }));
86+
87+
process::add( coroutine::add( COROUTINE(){
88+
coBegin
89+
90+
while( true ){
91+
while( process::size() > MAX_BATCH ){ coDelay(1000); }
92+
93+
worker::add([=](){ client();
94+
process::wait(); return -1; });
95+
96+
coNext; }
97+
98+
coFinish
99+
}));
100+
101+
}
102+
103+
/*────────────────────────────────────────────────────────────────────────────*/
104+
// BUGFIXED

0 commit comments

Comments
 (0)