From 9a13785fa5ddb669e6e9dc82c4f4ef487d1173ab Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?C=C3=A9dric=20Moreau?= <cem.moreau@gmail.com>
Date: Fri, 4 May 2018 17:39:22 +0200
Subject: [PATCH] [enh] Refactoring: type FileDAL.peerDAL
---
app/lib/dal/fileDAL.ts | 9 +-
app/lib/dal/indexDAL/abstract/PeerDAO.ts | 61 ++++++++++++
.../indexDAL/loki/LokiCollectionManager.ts | 2 +-
app/lib/dal/indexDAL/loki/LokiPeer.ts | 94 +++++++++++++++++++
app/lib/dal/indexDAL/loki/LokiTypes.ts | 2 +
app/modules/crawler/index.ts | 4 +-
app/modules/crawler/lib/garbager.ts | 2 +-
app/modules/crawler/lib/sync.ts | 3 +
test/fast/modules/crawler/peers_garbaging.js | 4 +-
test/integration/v1.0-modules-api.js | 4 +-
10 files changed, 174 insertions(+), 11 deletions(-)
create mode 100644 app/lib/dal/indexDAL/abstract/PeerDAO.ts
create mode 100644 app/lib/dal/indexDAL/loki/LokiPeer.ts
diff --git a/app/lib/dal/fileDAL.ts b/app/lib/dal/fileDAL.ts
index ea94bc431..8902b0707 100644
--- a/app/lib/dal/fileDAL.ts
+++ b/app/lib/dal/fileDAL.ts
@@ -27,7 +27,7 @@ import {
IndexEntry,
SindexEntry
} from "../indexer"
-import {DBPeer, PeerDAL} from "./sqliteDAL/PeerDAL"
+import {DBPeer} from "./sqliteDAL/PeerDAL"
import {TransactionDTO} from "../dto/TransactionDTO"
import {CertDAL, DBCert} from "./sqliteDAL/CertDAL"
import {DBWallet} from "./sqliteDAL/WalletDAL"
@@ -64,6 +64,8 @@ import {TxsDAO} from "./indexDAL/abstract/TxsDAO"
import {LokiJsDriver} from "./drivers/LokiJsDriver"
import {WalletDAO} from "./indexDAL/abstract/WalletDAO"
import {LokiWallet} from "./indexDAL/loki/LokiWallet"
+import {PeerDAO} from "./indexDAL/abstract/PeerDAO"
+import {LokiPeer} from "./indexDAL/loki/LokiPeer"
const fs = require('fs')
const path = require('path')
@@ -100,7 +102,6 @@ export class FileDAL {
// SQLite DALs
metaDAL:MetaDAL
- peerDAL:PeerDAL
idtyDAL:IdentityDAL
certDAL:CertDAL
msDAL:MembershipDAL
@@ -108,6 +109,7 @@ export class FileDAL {
// New DAO entities
blockDAL:BlockchainDAO
txsDAL:TxsDAO
+ peerDAL:PeerDAO
walletDAL:WalletDAO
bindexDAL:BIndexDAO
mindexDAL:MIndexDAO
@@ -130,7 +132,6 @@ export class FileDAL {
this.powDAL = new PowDAL(this.rootPath, params.fs)
this.confDAL = new ConfDAL(this.rootPath, params.fs)
this.metaDAL = new (require('./sqliteDAL/MetaDAL').MetaDAL)(this.sqliteDriver);
- this.peerDAL = new (require('./sqliteDAL/PeerDAL').PeerDAL)(this.sqliteDriver);
this.fakeBlockDAL = new (require('./sqliteDAL/BlockDAL').BlockDAL)(this.sqliteDriver);
this.blockDAL = new LokiBlockchain(this.loki.getLokiInstance())
this.fakeTxsDAL = new (require('./sqliteDAL/TxsDAL').TxsDAL)(this.sqliteDriver);
@@ -139,6 +140,7 @@ export class FileDAL {
this.idtyDAL = new (require('./sqliteDAL/IdentityDAL').IdentityDAL)(this.sqliteDriver);
this.certDAL = new (require('./sqliteDAL/CertDAL').CertDAL)(this.sqliteDriver);
this.msDAL = new (require('./sqliteDAL/MembershipDAL').MembershipDAL)(this.sqliteDriver);
+ this.peerDAL = new LokiPeer(this.loki.getLokiInstance())
this.walletDAL = new LokiWallet(this.loki.getLokiInstance())
this.bindexDAL = new LokiBIndex(this.loki.getLokiInstance())
this.mindexDAL = new LokiMIndex(this.loki.getLokiInstance())
@@ -174,6 +176,7 @@ export class FileDAL {
const dals = [
this.blockDAL,
this.txsDAL,
+ this.peerDAL,
this.walletDAL,
this.bindexDAL,
this.mindexDAL,
diff --git a/app/lib/dal/indexDAL/abstract/PeerDAO.ts b/app/lib/dal/indexDAL/abstract/PeerDAO.ts
new file mode 100644
index 000000000..9e3423b43
--- /dev/null
+++ b/app/lib/dal/indexDAL/abstract/PeerDAO.ts
@@ -0,0 +1,61 @@
+import {Initiable} from "../../sqliteDAL/Initiable"
+import {DBPeer} from "../../sqliteDAL/PeerDAL"
+
+export interface PeerDAO extends Initiable {
+
+ /**
+ * Trigger the initialization of the DAO. Called when the underlying DB is ready.
+ */
+ triggerInit(): void
+
+ listAll(): Promise<DBPeer[]>
+
+ withUPStatus(): Promise<DBPeer[]>
+
+ /**
+ * Saves a wallet.
+ * @param {DBPeer} peer
+ * @returns {Promise<DBPeer>}
+ */
+ savePeer(peer:DBPeer): Promise<DBPeer>
+
+ /**
+ * Find a wallet based on conditions.
+ * @param {string} pubkey
+ * @returns {Promise<DBPeer>}
+ */
+ getPeer(pubkey:string): Promise<DBPeer>
+
+ /**
+ * Find all peers with at least one endpoint matching given parameter.
+ * @param {string} ep
+ * @returns {Promise<DBPeer[]>}
+ */
+ getPeersWithEndpointsLike(ep:string): Promise<DBPeer[]>
+
+ /**
+ * Make a batch insert.
+ * @param records The records to insert as a batch.
+ */
+ insertBatch(records:DBPeer[]): Promise<void>
+
+ /**
+ * Remove a peer by its pubkey.
+ * @param {string} pubkey
+ * @returns {Promise<void>}
+ */
+ removePeerByPubkey(pubkey:string): Promise<void>
+
+ /**
+ * Remove peers that were set down before a certain datetime.
+ * @param {number} thresholdTime
+ * @returns {Promise<void>}
+ */
+ removePeersDownBefore(thresholdTime:number): Promise<void>
+
+ /**
+ * Remove all the peers.
+ * @returns {Promise<void>}
+ */
+ removeAll(): Promise<void>
+}
diff --git a/app/lib/dal/indexDAL/loki/LokiCollectionManager.ts b/app/lib/dal/indexDAL/loki/LokiCollectionManager.ts
index ad040eed8..f922c0fcf 100644
--- a/app/lib/dal/indexDAL/loki/LokiCollectionManager.ts
+++ b/app/lib/dal/indexDAL/loki/LokiCollectionManager.ts
@@ -12,7 +12,7 @@ export abstract class LokiCollectionManager<T> {
public constructor(
protected loki:any,
- protected collectionName:'iindex'|'mindex'|'cindex'|'sindex'|'bindex'|'blockchain'|'txs'|'wallet',
+ protected collectionName:'iindex'|'mindex'|'cindex'|'sindex'|'bindex'|'blockchain'|'txs'|'wallet'|'peer',
protected indices: (keyof T)[]) {
this.collectionIsInitialized = new Promise<void>(res => this.resolveCollection = res)
}
diff --git a/app/lib/dal/indexDAL/loki/LokiPeer.ts b/app/lib/dal/indexDAL/loki/LokiPeer.ts
new file mode 100644
index 000000000..2614d0ac7
--- /dev/null
+++ b/app/lib/dal/indexDAL/loki/LokiPeer.ts
@@ -0,0 +1,94 @@
+import {LokiCollectionManager} from "./LokiCollectionManager"
+import {PeerDAO} from "../abstract/PeerDAO"
+import {DBPeer} from "../../sqliteDAL/PeerDAL"
+
+export class LokiPeer extends LokiCollectionManager<DBPeer> implements PeerDAO {
+
+ constructor(loki:any) {
+ super(loki, 'peer', ['pubkey'])
+ }
+
+ cleanCache(): void {
+ }
+
+ async listAll(): Promise<DBPeer[]> {
+ return this.collection
+ .find({})
+ }
+
+ async withUPStatus(): Promise<DBPeer[]> {
+ return this.collection
+ .find({ status: 'UP' })
+ }
+
+ async getPeer(pubkey: string): Promise<DBPeer> {
+ return this.collection
+ .find({ pubkey })[0]
+ }
+
+ async insertBatch(peers: DBPeer[]): Promise<void> {
+ for (const p of peers) {
+ this.collection.insert(p)
+ }
+ }
+
+ async savePeer(peer: DBPeer): Promise<DBPeer> {
+ let updated = false
+ this.collection
+ .chain()
+ .find({ pubkey: peer.pubkey })
+ .update(p => {
+ p.version = peer.version
+ p.currency = peer.currency
+ p.status = peer.status
+ p.statusTS = peer.statusTS
+ p.hash = peer.hash
+ p.first_down = peer.first_down
+ p.last_try = peer.last_try
+ p.pubkey = peer.pubkey
+ p.block = peer.block
+ p.signature = peer.signature
+ p.endpoints = peer.endpoints
+ p.raw = peer.raw
+ updated = true
+ })
+ if (!updated) {
+ await this.insertBatch([peer])
+ }
+ return peer
+ }
+
+ async removePeerByPubkey(pubkey:string): Promise<void> {
+ this.collection
+ .chain()
+ .find({ pubkey })
+ .remove()
+ }
+
+ async removePeersDownBefore(thresholdTime:number): Promise<void> {
+ this.collection
+ .chain()
+ .find({
+ $and: [
+ { first_down: { $lt: thresholdTime } },
+ { first_down: { $gt: 0 } },
+ ]
+ })
+ .remove()
+ }
+
+ async removeAll(): Promise<void> {
+ this.collection
+ .chain()
+ .find({})
+ .remove()
+ }
+
+ async getPeersWithEndpointsLike(ep: string): Promise<DBPeer[]> {
+ return this.collection
+ .chain()
+ .find({})
+ .where(p => p.endpoints.filter(ep => ep.indexOf(ep) !== -1).length > 0)
+ .data()
+ }
+}
\ No newline at end of file
diff --git a/app/lib/dal/indexDAL/loki/LokiTypes.ts b/app/lib/dal/indexDAL/loki/LokiTypes.ts
index f61fea8d9..26150bbe1 100644
--- a/app/lib/dal/indexDAL/loki/LokiTypes.ts
+++ b/app/lib/dal/indexDAL/loki/LokiTypes.ts
@@ -24,6 +24,8 @@ export interface LokiChainableFind<T> {
update(cb:(t:T) => void): LokiChainableFind<T>
+ where(filter:(t:T) => boolean): LokiChainableFind<T>
+
remove(): LokiChainableFind<T>
compoundsort(sort:((string|((string|boolean)[]))[])): LokiChainableFind<T>
diff --git a/app/modules/crawler/index.ts b/app/modules/crawler/index.ts
index 653466f47..ddf6fc1fe 100644
--- a/app/modules/crawler/index.ts
+++ b/app/modules/crawler/index.ts
@@ -159,7 +159,7 @@ export const CrawlerDependency = {
const toPort = params[4];
const logger = server.logger;
try {
- const peers = fromHost && fromPort ? [{ endpoints: [['BASIC_MERKLED_API', fromHost, fromPort].join(' ')] }] : await server.dal.peerDAL.query('SELECT * FROM peer WHERE status = ?', ['UP'])
+ const peers = fromHost && fromPort ? [{ endpoints: [['BASIC_MERKLED_API', fromHost, fromPort].join(' ')] }] : await server.dal.peerDAL.withUPStatus()
// Memberships
for (const p of peers) {
const peer = PeerDTO.fromJSONObject(p)
@@ -312,7 +312,7 @@ export const CrawlerDependency = {
const fromPort = params[3]
const logger = server.logger;
try {
- const peers = fromHost && fromPort ? [{ endpoints: [['BASIC_MERKLED_API', fromHost, fromPort].join(' ')] }] : await server.dal.peerDAL.query('SELECT * FROM peer WHERE status = ?', ['UP'])
+ const peers = fromHost && fromPort ? [{ endpoints: [['BASIC_MERKLED_API', fromHost, fromPort].join(' ')] }] : await server.dal.peerDAL.withUPStatus()
// Memberships
for (const p of peers) {
const peer = PeerDTO.fromJSONObject(p)
diff --git a/app/modules/crawler/lib/garbager.ts b/app/modules/crawler/lib/garbager.ts
index f748a63a6..852d33187 100644
--- a/app/modules/crawler/lib/garbager.ts
+++ b/app/modules/crawler/lib/garbager.ts
@@ -16,5 +16,5 @@ import {Server} from "../../../../server"
export const cleanLongDownPeers = async (server:Server, now:number) => {
const first_down_limit = now - CrawlerConstants.PEER_LONG_DOWN * 1000;
- await server.dal.peerDAL.query('DELETE FROM peer WHERE first_down < ' + first_down_limit)
+ await server.dal.peerDAL.removePeersDownBefore(first_down_limit)
}
diff --git a/app/modules/crawler/lib/sync.ts b/app/modules/crawler/lib/sync.ts
index 424b89010..239be33cd 100644
--- a/app/modules/crawler/lib/sync.ts
+++ b/app/modules/crawler/lib/sync.ts
@@ -328,6 +328,9 @@ export class Synchroniser extends stream.Duplex {
//=======
await this.syncPeers(nopeers, fullSync, this.host, this.port, to)
+ // Trim the loki data
+ await this.server.dal.loki.flushAndTrimData()
+
this.watcher.end();
this.push({ sync: true });
this.logger.info('Sync finished.');
diff --git a/test/fast/modules/crawler/peers_garbaging.js b/test/fast/modules/crawler/peers_garbaging.js
index 4d44184db..9e29d392a 100644
--- a/test/fast/modules/crawler/peers_garbaging.js
+++ b/test/fast/modules/crawler/peers_garbaging.js
@@ -38,10 +38,10 @@ describe('Peers garbaging', () => {
yield server.dal.peerDAL.savePeer({ pubkey: 'B', version: 1, currency: 'c', first_down: 1484827199999, statusTS: 1485000000000, block: '2393-H' });
yield server.dal.peerDAL.savePeer({ pubkey: 'C', version: 1, currency: 'c', first_down: 1484827200000, statusTS: 1485000000000, block: '2393-H' });
yield server.dal.peerDAL.savePeer({ pubkey: 'D', version: 1, currency: 'c', first_down: 1484820000000, statusTS: 1485000000000, block: '2393-H' });
- (yield server.dal.peerDAL.sqlListAll()).should.have.length(4);
+ (yield server.dal.peerDAL.listAll()).should.have.length(4);
const now = 1485000000000;
yield garbager.cleanLongDownPeers(server, now);
- (yield server.dal.peerDAL.sqlListAll()).should.have.length(2);
+ (yield server.dal.peerDAL.listAll()).should.have.length(2);
})
}]
}
diff --git a/test/integration/v1.0-modules-api.js b/test/integration/v1.0-modules-api.js
index 9c6b3b76b..1cee0379a 100644
--- a/test/integration/v1.0-modules-api.js
+++ b/test/integration/v1.0-modules-api.js
@@ -28,7 +28,7 @@ describe("v1.0 Module API", () => {
it('should be able to execute `hello` command with quickRun', () => co(function*() {
duniter.statics.setOnRunDone(() => { /* Do not exit the process */ })
const absolutePath = path.join(__dirname, './scenarios/hello-plugin.js')
- process.argv = ['', absolutePath, 'hello-world']
+ process.argv = ['', absolutePath, 'hello-world', '--memory']
const res = yield duniter.statics.quickRun(absolutePath)
res.should.equal('Hello world! from within Duniter.')
}))
@@ -111,7 +111,7 @@ describe("v1.0 Module API", () => {
cli: [{
name: 'gimme-conf',
desc: 'Returns the configuration object.',
- onDatabaseExecute: (server, conf, program, params, startServices, stopServices) => co(function*() {
+ onConfiguredExecute: (server, conf, program, params, startServices, stopServices) => co(function*() {
// Gimme the conf!
return conf;
})
--
GitLab