Artifact
1e0198fa610dfca4069956285ab338f79692ff32dcf751fb7c717c960d4eec17:
0000: 65 78 74 65 72 6e 20 63 72 61 74 65 20 63 6f 6e extern crate con
0010: 66 69 67 3b 0a 65 78 74 65 72 6e 20 63 72 61 74 fig;.extern crat
0020: 65 20 70 6f 73 74 67 72 65 73 3b 0a 65 78 74 65 e postgres;.exte
0030: 72 6e 20 63 72 61 74 65 20 72 65 64 69 73 3b 0a rn crate redis;.
0040: 65 78 74 65 72 6e 20 63 72 61 74 65 20 72 65 67 extern crate reg
0050: 65 78 3b 0a 0a 75 73 65 20 70 6f 73 74 67 72 65 ex;..use postgre
0060: 73 3a 3a 7b 43 6f 6e 6e 65 63 74 69 6f 6e 2c 20 s::{Connection,
0070: 54 6c 73 4d 6f 64 65 7d 3b 0a 75 73 65 20 72 65 TlsMode};.use re
0080: 64 69 73 3a 3a 43 6f 6d 6d 61 6e 64 73 3b 0a 75 dis::Commands;.u
0090: 73 65 20 72 65 67 65 78 3a 3a 52 65 67 65 78 3b se regex::Regex;
00a0: 0a 75 73 65 20 73 74 64 3a 3a 63 6f 6c 6c 65 63 .use std::collec
00b0: 74 69 6f 6e 73 3a 3a 48 61 73 68 4d 61 70 3b 0a tions::HashMap;.
00c0: 75 73 65 20 73 74 64 3a 3a 74 69 6d 65 3a 3a 44 use std::time::D
00d0: 75 72 61 74 69 6f 6e 3b 0a 75 73 65 20 73 74 64 uration;.use std
00e0: 3a 3a 74 68 72 65 61 64 3b 0a 0a 66 6e 20 6d 61 ::thread;..fn ma
00f0: 69 6e 28 29 20 7b 0a 20 20 20 20 6c 6f 6f 70 20 in() {. loop
0100: 7b 0a 20 20 20 20 20 20 20 20 6c 65 74 20 6d 75 {. let mu
0110: 74 20 73 65 74 74 69 6e 67 73 20 3d 20 63 6f 6e t settings = con
0120: 66 69 67 3a 3a 43 6f 6e 66 69 67 3a 3a 64 65 66 fig::Config::def
0130: 61 75 6c 74 28 29 3b 0a 20 20 20 20 20 20 20 20 ault();.
0140: 73 65 74 74 69 6e 67 73 2e 6d 65 72 67 65 28 63 settings.merge(c
0150: 6f 6e 66 69 67 3a 3a 46 69 6c 65 3a 3a 77 69 74 onfig::File::wit
0160: 68 5f 6e 61 6d 65 28 22 6f 66 66 6c 6f 61 64 22 h_name("offload"
0170: 29 29 2e 65 78 70 65 63 74 28 22 43 61 6e 27 74 )).expect("Can't
0180: 20 72 65 61 64 20 63 6f 6e 66 69 67 75 72 61 74 read configurat
0190: 69 6f 6e 20 66 69 6c 65 22 29 3b 0a 0a 20 20 20 ion file");..
01a0: 20 20 20 20 20 6c 65 74 20 72 65 64 64 62 20 3d let reddb =
01b0: 20 72 65 64 69 73 3a 3a 43 6c 69 65 6e 74 3a 3a redis::Client::
01c0: 6f 70 65 6e 28 22 72 65 64 69 73 3a 2f 2f 31 32 open("redis://12
01d0: 37 2e 30 2e 30 2e 31 2f 22 29 2e 65 78 70 65 63 7.0.0.1/").expec
01e0: 74 28 22 43 61 6e 27 74 20 63 6f 6e 6e 65 63 74 t("Can't connect
01f0: 20 74 6f 20 74 68 65 20 64 61 74 61 62 61 73 65 to the database
0200: 22 29 3b 0a 20 20 20 20 20 20 20 20 6c 65 74 20 ");. let
0210: 72 65 64 20 3d 20 72 65 64 64 62 2e 67 65 74 5f red = reddb.get_
0220: 63 6f 6e 6e 65 63 74 69 6f 6e 28 29 2e 65 78 70 connection().exp
0230: 65 63 74 28 22 43 61 6e 27 74 20 69 6e 69 74 69 ect("Can't initi
0240: 61 6c 69 7a 65 20 6e 65 77 20 63 6f 6e 6e 65 63 alize new connec
0250: 74 69 6f 6e 22 29 3b 0a 0a 20 20 20 20 20 20 20 tion");..
0260: 20 6c 65 74 20 72 65 20 3d 20 52 65 67 65 78 3a let re = Regex:
0270: 3a 6e 65 77 28 22 5e 28 5b 30 2d 39 41 2d 5a 5d :new("^([0-9A-Z]
0280: 2b 29 5f 28 5b 61 2d 7a 2e 2f 30 2d 39 2d 5d 2b +)_([a-z./0-9-]+
0290: 29 5f 28 5b 30 2d 39 2e 5d 2b 29 24 22 29 2e 65 )_([0-9.]+)$").e
02a0: 78 70 65 63 74 28 22 43 61 6e 27 74 20 70 61 72 xpect("Can't par
02b0: 73 65 20 72 65 67 65 78 70 22 29 3b 0a 20 20 20 se regexp");.
02c0: 20 20 20 20 20 6c 65 74 20 63 6f 6e 6e 20 3d 20 let conn =
02d0: 43 6f 6e 6e 65 63 74 69 6f 6e 3a 3a 63 6f 6e 6e Connection::conn
02e0: 65 63 74 28 73 65 74 74 69 6e 67 73 2e 67 65 74 ect(settings.get
02f0: 5f 73 74 72 28 22 70 67 22 29 2e 65 78 70 65 63 _str("pg").expec
0300: 74 28 22 50 6f 73 74 67 72 65 73 20 63 6f 6e 6e t("Postgres conn
0310: 65 63 74 69 6f 6e 20 61 62 73 65 6e 74 20 69 6e ection absent in
0320: 20 63 6f 6e 66 69 67 22 29 2c 20 54 6c 73 4d 6f config"), TlsMo
0330: 64 65 3a 3a 4e 6f 6e 65 29 2e 65 78 70 65 63 74 de::None).expect
0340: 28 22 43 61 6e 27 74 20 63 6f 6e 6e 65 63 74 20 ("Can't connect
0350: 74 6f 20 70 6f 73 74 67 72 65 73 22 29 3b 0a 0a to postgres");..
0360: 20 20 20 20 20 20 20 20 6c 65 74 20 73 63 68 65 let sche
0370: 6d 61 73 20 3d 20 73 65 74 74 69 6e 67 73 2e 67 mas = settings.g
0380: 65 74 5f 61 72 72 61 79 28 22 73 63 68 65 6d 61 et_array("schema
0390: 73 22 29 2e 65 78 70 65 63 74 28 22 53 63 68 65 s").expect("Sche
03a0: 6d 61 20 6c 69 73 74 20 6e 6f 74 20 66 6f 75 6e ma list not foun
03b0: 64 20 69 6e 20 63 6f 6e 66 69 67 22 29 2e 69 6e d in config").in
03c0: 74 6f 5f 69 74 65 72 28 29 2e 6d 61 70 28 7c 76 to_iter().map(|v
03d0: 61 6c 75 65 7c 20 63 6f 6e 66 69 67 3a 3a 56 61 alue| config::Va
03e0: 6c 75 65 3a 3a 69 6e 74 6f 5f 73 74 72 28 76 61 lue::into_str(va
03f0: 6c 75 65 29 2e 65 78 70 65 63 74 28 22 57 65 20 lue).expect("We
0400: 72 65 71 75 69 72 65 20 73 74 72 69 6e 67 20 68 require string h
0410: 65 72 65 22 29 29 3b 0a 20 20 20 20 20 20 20 20 ere"));.
0420: 66 6f 72 20 73 63 68 65 6d 61 20 69 6e 20 73 63 for schema in sc
0430: 68 65 6d 61 73 20 7b 0a 20 20 20 20 20 20 20 20 hemas {.
0440: 20 20 20 20 6c 65 74 20 64 61 74 61 5f 6b 65 79 let data_key
0450: 20 3d 20 73 63 68 65 6d 61 2e 74 6f 5f 6f 77 6e = schema.to_own
0460: 65 64 28 29 20 2b 20 22 5f 63 6f 75 6e 74 65 72 ed() + "_counter
0470: 5f 70 65 6e 64 69 6e 67 22 3b 0a 20 20 20 20 20 _pending";.
0480: 20 20 20 20 20 20 20 6c 65 74 20 63 61 63 68 65 let cache
0490: 5f 6b 65 79 20 3d 20 73 63 68 65 6d 61 2e 74 6f _key = schema.to
04a0: 5f 6f 77 6e 65 64 28 29 20 2b 20 22 5f 63 6f 75 _owned() + "_cou
04b0: 6e 74 65 72 5f 70 65 6e 64 69 6e 67 5f 6e 6f 77 nter_pending_now
04c0: 22 3b 0a 0a 20 20 20 20 20 20 20 20 20 20 20 20 ";..
04d0: 69 66 20 72 65 64 2e 65 78 69 73 74 73 28 26 63 if red.exists(&c
04e0: 61 63 68 65 5f 6b 65 79 29 2e 65 78 70 65 63 74 ache_key).expect
04f0: 28 22 43 61 6e 27 74 20 71 75 65 72 79 20 63 61 ("Can't query ca
0500: 63 68 65 20 65 78 69 73 74 61 6e 63 65 22 29 20 che existance")
0510: 7b 0a 20 20 20 20 20 20 20 20 20 20 20 20 7d 20 {. }
0520: 65 6c 73 65 20 69 66 20 72 65 64 2e 65 78 69 73 else if red.exis
0530: 74 73 28 26 64 61 74 61 5f 6b 65 79 29 2e 65 78 ts(&data_key).ex
0540: 70 65 63 74 28 22 43 61 6e 27 74 20 71 75 65 72 pect("Can't quer
0550: 79 20 64 61 74 61 20 65 78 69 73 74 61 6e 63 65 y data existance
0560: 22 29 20 7b 0a 20 20 20 20 20 20 20 20 20 20 20 ") {.
0570: 20 20 20 20 20 6c 65 74 20 5f 20 3a 20 62 6f 6f let _ : boo
0580: 6c 20 3d 20 72 65 64 2e 72 65 6e 61 6d 65 5f 6e l = red.rename_n
0590: 78 28 26 64 61 74 61 5f 6b 65 79 2c 20 26 63 61 x(&data_key, &ca
05a0: 63 68 65 5f 6b 65 79 29 2e 65 78 70 65 63 74 28 che_key).expect(
05b0: 22 43 61 6e 27 74 20 72 65 61 64 64 20 75 6e 66 "Can't readd unf
05c0: 6c 75 73 68 65 64 20 68 61 73 68 22 29 3b 0a 20 lushed hash");.
05d0: 20 20 20 20 20 20 20 20 20 20 20 7d 20 65 6c 73 } els
05e0: 65 20 7b 0a 20 20 20 20 20 20 20 20 20 20 20 20 e {.
05f0: 20 20 20 20 63 6f 6e 74 69 6e 75 65 3b 0a 20 20 continue;.
0600: 20 20 20 20 20 20 20 20 20 20 7d 0a 0a 20 20 20 }..
0610: 20 20 20 20 20 20 20 20 20 6c 65 74 20 73 74 61 let sta
0620: 74 73 20 3a 20 48 61 73 68 4d 61 70 3c 53 74 72 ts : HashMap<Str
0630: 69 6e 67 2c 20 69 31 36 3e 20 3d 20 72 65 64 2e ing, i16> = red.
0640: 68 67 65 74 61 6c 6c 28 26 63 61 63 68 65 5f 6b hgetall(&cache_k
0650: 65 79 29 2e 65 78 70 65 63 74 28 22 43 61 6e 27 ey).expect("Can'
0660: 74 20 71 75 65 72 79 20 66 6f 72 20 73 74 6f 72 t query for stor
0670: 65 64 20 73 74 61 74 20 6b 65 79 73 22 29 3b 0a ed stat keys");.
0680: 0a 20 20 20 20 20 20 20 20 20 20 20 20 2f 2f 70 . //p
0690: 72 69 6e 74 6c 6e 21 28 22 23 20 7b 3a 3f 7d 22 rintln!("# {:?}"
06a0: 2c 20 73 74 61 74 73 29 3b 0a 0a 20 20 20 20 20 , stats);..
06b0: 20 20 20 20 20 20 20 6c 65 74 20 74 72 61 6e 73 let trans
06c0: 20 3d 20 63 6f 6e 6e 2e 74 72 61 6e 73 61 63 74 = conn.transact
06d0: 69 6f 6e 28 29 2e 65 78 70 65 63 74 28 22 43 61 ion().expect("Ca
06e0: 6e 27 74 20 73 74 61 72 74 20 74 72 61 6e 73 61 n't start transa
06f0: 63 74 69 6f 6e 22 29 3b 0a 20 20 20 20 20 20 20 ction");.
0700: 20 20 20 20 20 6c 65 74 20 73 74 6d 74 20 3d 20 let stmt =
0710: 74 72 61 6e 73 2e 70 72 65 70 61 72 65 28 26 66 trans.prepare(&f
0720: 6f 72 6d 61 74 21 28 22 73 65 6c 65 63 74 20 7b ormat!("select {
0730: 7d 2e 6d 65 72 67 65 5f 63 6f 75 6e 74 65 72 28 }.merge_counter(
0740: 24 31 3a 3a 74 65 78 74 2c 20 24 32 3a 3a 74 65 $1::text, $2::te
0750: 78 74 2c 20 28 24 33 3a 3a 74 65 78 74 29 3a 3a xt, ($3::text)::
0760: 69 6e 65 74 2c 20 24 34 3a 3a 73 6d 61 6c 6c 69 inet, $4::smalli
0770: 6e 74 29 3b 22 2c 20 26 73 63 68 65 6d 61 29 29 nt);", &schema))
0780: 2e 65 78 70 65 63 74 28 22 43 61 6e 27 74 20 70 .expect("Can't p
0790: 72 65 70 61 72 65 20 73 74 61 74 65 6d 65 6e 74 repare statement
07a0: 22 29 3b 0a 20 20 20 20 20 20 20 20 20 20 20 20 ");.
07b0: 66 6f 72 20 28 63 6c 69 65 6e 74 2c 20 63 6f 75 for (client, cou
07c0: 6e 74 29 20 69 6e 20 73 74 61 74 73 20 7b 0a 20 nt) in stats {.
07d0: 20 20 20 20 20 20 20 20 20 20 20 20 20 20 20 2f /
07e0: 2f 70 72 69 6e 74 6c 6e 21 28 22 23 20 7b 3a 3f /println!("# {:?
07f0: 7d 3a 20 7b 3a 3f 7d 22 2c 20 26 63 6c 69 65 6e }: {:?}", &clien
0800: 74 2c 20 26 63 6f 75 6e 74 29 3b 0a 20 20 20 20 t, &count);.
0810: 20 20 20 20 20 20 20 20 20 20 20 20 6c 65 74 20 let
0820: 63 61 70 20 3d 20 72 65 2e 63 61 70 74 75 72 65 cap = re.capture
0830: 73 28 26 63 6c 69 65 6e 74 29 2e 65 78 70 65 63 s(&client).expec
0840: 74 28 22 43 6c 69 65 6e 74 20 6d 61 74 63 68 20 t("Client match
0850: 66 61 69 6c 65 64 22 29 3b 0a 20 20 20 20 20 20 failed");.
0860: 20 20 20 20 20 20 20 20 20 20 2f 2f 6c 65 74 20 //let
0870: 61 64 64 72 20 3d 20 49 70 41 64 64 72 3a 3a 56 addr = IpAddr::V
0880: 34 28 63 61 70 5b 33 5d 2e 70 61 72 73 65 28 29 4(cap[3].parse()
0890: 2e 65 78 70 65 63 74 28 22 43 61 6e 27 74 20 70 .expect("Can't p
08a0: 61 72 73 65 20 49 50 22 29 29 3b 0a 20 20 20 20 arse IP"));.
08b0: 20 20 20 20 20 20 20 20 20 20 20 20 2f 2f 70 72 //pr
08c0: 69 6e 74 6c 6e 21 28 22 69 6e 73 65 72 74 20 69 intln!("insert i
08d0: 6e 74 6f 20 78 20 76 61 6c 75 65 73 28 7b 3a 3f nto x values({:?
08e0: 7d 2c 20 7b 3a 3f 7d 2c 20 7b 3a 3f 7d 2c 20 7b }, {:?}, {:?}, {
08f0: 3a 3f 7d 29 3b 22 2c 20 26 63 61 70 5b 31 5d 2c :?});", &cap[1],
0900: 20 26 63 61 70 5b 32 5d 2c 20 26 63 61 70 5b 33 &cap[2], &cap[3
0910: 5d 2c 20 26 63 6f 75 6e 74 29 3b 0a 20 20 20 20 ], &count);.
0920: 20 20 20 20 20 20 20 20 20 20 20 20 73 74 6d 74 stmt
0930: 2e 65 78 65 63 75 74 65 28 26 5b 26 63 61 70 5b .execute(&[&cap[
0940: 31 5d 2e 74 6f 5f 73 74 72 69 6e 67 28 29 2c 20 1].to_string(),
0950: 26 63 61 70 5b 32 5d 2e 74 6f 5f 73 74 72 69 6e &cap[2].to_strin
0960: 67 28 29 2c 20 26 63 61 70 5b 33 5d 2e 74 6f 5f g(), &cap[3].to_
0970: 73 74 72 69 6e 67 28 29 2c 20 26 63 6f 75 6e 74 string(), &count
0980: 5d 29 2e 65 78 70 65 63 74 28 22 43 61 6e 27 74 ]).expect("Can't
0990: 20 65 78 65 63 75 74 65 20 70 72 65 70 61 72 65 execute prepare
09a0: 64 22 29 3b 0a 20 20 20 20 20 20 20 20 20 20 20 d");.
09b0: 20 20 20 20 20 2f 2f 74 72 61 6e 73 2e 71 75 65 //trans.que
09c0: 72 79 28 26 66 6f 72 6d 61 74 21 28 22 73 65 6c ry(&format!("sel
09d0: 65 63 74 20 7b 7d 2e 6d 65 72 67 65 5f 63 6f 75 ect {}.merge_cou
09e0: 6e 74 65 72 28 24 31 3a 3a 74 65 78 74 2c 20 24 nter($1::text, $
09f0: 32 3a 3a 74 65 78 74 2c 20 24 33 3a 3a 69 6e 65 2::text, $3::ine
0a00: 74 2c 20 24 34 3a 3a 73 6d 61 6c 6c 69 6e 74 29 t, $4::smallint)
0a10: 3b 22 2c 20 26 73 63 68 65 6d 61 29 2c 20 26 5b ;", &schema), &[
0a20: 26 63 61 70 5b 31 5d 2e 74 6f 5f 73 74 72 69 6e &cap[1].to_strin
0a30: 67 28 29 2c 20 26 63 61 70 5b 32 5d 2e 74 6f 5f g(), &cap[2].to_
0a40: 73 74 72 69 6e 67 28 29 2c 20 26 63 61 70 5b 33 string(), &cap[3
0a50: 5d 2e 74 6f 5f 73 74 72 69 6e 67 28 29 2c 20 26 ].to_string(), &
0a60: 63 6f 75 6e 74 5d 29 2e 65 78 70 65 63 74 28 22 count]).expect("
0a70: 43 61 6e 27 74 20 70 72 65 70 61 72 65 20 73 74 Can't prepare st
0a80: 61 74 65 6d 65 6e 74 22 29 3b 0a 20 20 20 20 20 atement");.
0a90: 20 20 20 20 20 20 20 7d 0a 20 20 20 20 20 20 20 }.
0aa0: 20 20 20 20 20 74 72 61 6e 73 2e 63 6f 6d 6d 69 trans.commi
0ab0: 74 28 29 2e 65 78 70 65 63 74 28 22 43 61 6e 27 t().expect("Can'
0ac0: 74 20 63 6f 6d 6d 69 74 20 74 72 61 6e 73 61 63 t commit transac
0ad0: 74 69 6f 6e 22 29 3b 0a 20 20 20 20 20 20 20 20 tion");.
0ae0: 20 20 20 20 72 65 64 2e 64 65 6c 28 63 61 63 68 red.del(cach
0af0: 65 5f 6b 65 79 29 2e 65 78 70 65 63 74 28 22 43 e_key).expect("C
0b00: 61 6e 27 74 20 72 65 6d 6f 76 65 20 73 74 61 6c an't remove stal
0b10: 65 20 6b 65 79 22 29 0a 20 20 20 20 20 20 20 20 e key").
0b20: 7d 0a 20 20 20 20 20 20 20 20 74 68 72 65 61 64 }. thread
0b30: 3a 3a 73 6c 65 65 70 28 44 75 72 61 74 69 6f 6e ::sleep(Duration
0b40: 3a 3a 6e 65 77 28 73 65 74 74 69 6e 67 73 2e 67 ::new(settings.g
0b50: 65 74 28 22 64 65 6c 61 79 22 29 2e 65 78 70 65 et("delay").expe
0b60: 63 74 28 22 44 65 6c 61 79 20 73 70 65 63 69 66 ct("Delay specif
0b70: 69 63 61 74 69 6f 6e 20 61 62 73 65 6e 74 20 69 ication absent i
0b80: 6e 20 63 6f 6e 66 69 67 22 29 2c 20 30 29 29 3b n config"), 0));
0b90: 0a 20 20 20 20 7d 0a 7d 0a . }.}.