48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
|
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
|
-
+
|
.connect_timeout(std::time::Duration::new(300, 0))
.idle_timeout(std::time::Duration::new(60, 0))
.connect_lazy(&settings.get_str("pg")?)?,
};
let clone = core.clone();
tokio::spawn(async move {
if let Err(err) = &clone.autofetch().await {
if let Err(err) = clone.debug(&err.to_string()) {
if let Err(err) = clone.debug(&format!("{:?}", err)) {
eprintln!("Autofetch error: {}", err);
};
}
});
Ok(core)
}
|
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
|
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
|
-
+
-
+
|
let id: i64 = id.into();
let mut conn = self.pool.acquire().await
.with_context(|| format!("🛑 Enable fetch conn:\n{:?}", &self.pool))?;
match sqlx::query("update rsstg_source set enabled = true where source_id = $1 and owner = $2")
.bind(source_id)
.bind(id)
.execute(&mut conn).await
.with_context(|| format!("🛑 Enable source:\n\n{:?}", &self.pool))?
.with_context(|| format!("🛑 Enable source:\n{:?}", &self.pool))?
.rows_affected() {
1 => { Ok("Source disabled\\.") },
0 => { Ok("Source not found\\.") },
_ => { Err(anyhow!("Database error.")) },
}
}
async fn disable<S>(&self, source_id: &i32, id: S) -> Result<&str>
where S: Into<i64> {
let id: i64 = id.into();
let mut conn = self.pool.acquire().await
.with_context(|| format!("🛑 Disable fetch conn:\n{:?}", &self.pool))?;
match sqlx::query("update rsstg_source set enabled = false where source_id = $1 and owner = $2")
.bind(source_id)
.bind(id)
.execute(&mut conn).await
.with_context(|| format!("🛑 Disable source:\n\n{:?}", &self.pool))?
.with_context(|| format!("🛑 Disable source:\n{:?}", &self.pool))?
.rows_affected() {
1 => { Ok("Source disabled\\.") },
0 => { Ok("Source not found\\.") },
_ => { Err(anyhow!("Database error.")) },
}
}
|
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
|
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
|
-
+
|
sqlx::query("update rsstg_source set last_scrape = now() + interval '1 hour' where source_id = $1;")
.bind(source_id)
.execute(&mut conn).await
.with_context(|| format!("🛑 Lock source:\n\n{:?}", &self.pool))?;
let clone = self.clone();
tokio::spawn(async move {
if let Err(err) = clone.check(&source_id, owner, true).await {
if let Err(err) = clone.debug(&err.to_string()) {
if let Err(err) = clone.debug(&format!("{:?}", err)) {
eprintln!("Check error: {}", err);
};
};
});
} else {
if next_fetch - now < delay {
delay = next_fetch - now;
|
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
|
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
|
-
+
|
let core = Core::new(settings).await?;
let mut stream = core.stream();
while let Some(update) = stream.next().await {
if let Err(err) = handle(update?, &core).await {
core.debug(&err.to_string())?;
core.debug(&format!("{:?}", err))?;
};
}
Ok(())
}
async fn handle(update: telegram_bot::Update, core: &Core) -> Result<()> {
|