wn/MqttDispatcher
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

remove leading underscore, fix next call in mongosave

Browse Source
This commit is contained in:
Wolfgang Hottgenroth
2017-08-24 15:51:34 +02:00
parent a0a922f851
commit 25116aae89
14 changed files with 230 additions and 170 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
src/callchain.ts
View File

@ -13,29 +13,29 @@ class LastChainItem implements Receivable {
}
public send(message: any) {
log.info(`Last chain item, final result ${message}`)
log.info(`Last chain item, final result ${JSON.stringify(message)}`)
}
}
let lastChainItem : LastChainItem = new LastChainItem();
export abstract class AChainItem extends events.EventEmitter implements Receivable {
protected _label : string
protected _next : Receivable
protected label : string
protected next : Receivable
constructor(label : string) {
super()
this._label = label
this._next = lastChainItem
this.label = label
this.next = lastChainItem
}
public toString() : string {
return `<${this._label}>`
return `<${this.label}>`
}
public registerNext(next : AChainItem) :void {
this._next = next
this.next = next
}
public send(message : any) : void {
@ -45,6 +45,24 @@ export abstract class AChainItem extends events.EventEmitter implements Receivab
public abstract begin() : void
}
export abstract class AAsyncBaseChainItem extends AChainItem {
constructor(label : string) {
super(label)
}
protected abstract func(message : any, finished : (message : any) => void) : void;
public begin() :void {
if (this.next != null) {
this.next.begin()
}
this.addListener('yourturn', (message : any) : void => {
log.info(`Calling ${this.toString()}`)
this.func(message, this.next.send)
})
}
}
export abstract class ABaseChainItem extends AChainItem {
constructor(label : string) {
super(label)
@ -53,31 +71,31 @@ export abstract class ABaseChainItem extends AChainItem {
protected abstract func(message : any) : any;
public begin() :void {
if (this._next != null) {
this._next.begin()
if (this.next != null) {
this.next.begin()
}
this.addListener('yourturn', (message : any) : void => {
log.info(`Calling ${this.toString()}`)
let result : any = this.func(message)
this._next.send(result)
this.next.send(result)
})
}
}
export class ChainItem extends ABaseChainItem {
private _chainItemFunc : ChainItemFunc
private chainItemFunc : ChainItemFunc
public toString() : string {
let funcName : string = (this._chainItemFunc.name === "") ? "lambda" : this._chainItemFunc.name
return `<${funcName}, ${this._label}>`
let funcName : string = (this.chainItemFunc.name === "") ? "lambda" : this.chainItemFunc.name
return `<${funcName}, ${this.label}>`
}
public registerFunc(func : ChainItemFunc) : void {
this._chainItemFunc = func
this.chainItemFunc = func
}
protected func(message : any) : any {
return this._chainItemFunc(message)
return this.chainItemFunc(message)
}
}

51
src/espthermtojson.ts
View File

@ -1,17 +1,52 @@
import * as log from './log'
import * as utils from './utils'
export class EspThermData {
private temperature : number
private voltage : number
private timeConsumed : number
constructor(temperature:number, voltage:number, timeConsumed:number) {
this.temperature = temperature
this.voltage = voltage
this.timeConsumed = timeConsumed
}
toString() :string {
return JSON.stringify(this)
}
toJSON() : any {
return utils.jsonPrepaper(this, [])
}
}
export class EspThermMetadata {
private client : string
private timestamp : Date
constructor(client : string, timestamp? : Date) {
this.client = client
this.timestamp = (timestamp) ? timestamp : new Date()
}
toString() :string {
return JSON.stringify(this)
}
toJSON() : any {
return utils.jsonPrepaper(this, [])
}
}
export class EspThermMessage {
private _client : string
private _temperature : number
private _voltage : number
private _timeConsumed : number
private data : EspThermData
private metadata : EspThermMetadata
constructor(client:string, temperature:number, voltage:number, timeConsumed:number) {
this._client = client
this._temperature = temperature
this._voltage = voltage
this._timeConsumed = timeConsumed
this.data = new EspThermData(temperature, voltage, timeConsumed)
this.metadata = new EspThermMetadata(client)
}
toString() :string {

45
src/mongosave.ts
View File

@ -3,51 +3,60 @@ import * as log from './log'
import * as MongoDB from 'mongodb'
export class MongoSave extends CallChain.ABaseChainItem {
private _url : string
private _mongoClient : MongoDB.MongoClient
private _dbh : MongoDB.Db | undefined
private _connectPending : boolean
export interface MongoItem {
id : MongoDB.ObjectID
payload : any
}
export class MongoSave extends CallChain.AAsyncBaseChainItem {
private url : string
private mongoClient : MongoDB.MongoClient
private dbh : MongoDB.Db | undefined
private connectPending : boolean
constructor(url:string) {
super('MongoSave')
this._url = url
this._mongoClient = new MongoDB.MongoClient()
this._connectPending = false
this.url = url
this.mongoClient = new MongoDB.MongoClient()
this.connectPending = false
}
protected func(message : any) : any {
if (! this._dbh) {
protected func(message : any, finished : (message : any) => void) : void {
if (! this.dbh) {
log.info("Not database connection yet")
if (! this._connectPending) {
this._connectPending = true
this._mongoClient.connect(this._url)
if (! this.connectPending) {
this.connectPending = true
this.mongoClient.connect(this.url)
.then((db:MongoDB.Db) => {
log.info("Successfully opened MongoDB connect")
this._dbh = db
this.dbh = db
})
.catch((err) => {
log.error(`Failure when opening MongoDB connect: ${err}`)
this._dbh = undefined
this.dbh = undefined
})
} else {
log.info("Connecting to database is pending")
}
}
if (this._dbh) {
if (this.dbh) {
log.info("Database handle is available")
let coll : MongoDB.Collection = this._dbh.collection("iot")
let coll : MongoDB.Collection = this.dbh.collection("iot")
coll.insertOne(message)
.then((res : MongoDB.InsertOneWriteOpResult) => {
log.info(`Successfully wrote one item in database: ${res.insertedId}`)
let nextValue = <MongoItem>{id: res.insertedId, payload: message}
finished(nextValue)
})
.catch((err : any) => {
log.error(`Failure when trying to write one item in database: ${err}`)
log.error("Chain interrupted")
})
} else {
log.error(`No database connection yet, drop message ${message}`)
log.error("Chain interrupted")
}
return "<<" + message + ">>"
log.info(`Returning from ${this.label}`)
}
/*

42
src/mqttdispatcher.ts
View File

@ -25,24 +25,24 @@ export interface IDispatcher {
}
export class MqttDispatcher implements IDispatcher {
private _mqttClient: Mqtt.Client
private _mqttOptions: Mqtt.IClientOptions = {}
private _mqttBrokerUrl: string
private _topicHandlers: TopicHandler[]
private mqttClient: Mqtt.Client
private mqttOptions: Mqtt.IClientOptions = {}
private mqttBrokerUrl: string
private topicHandlers: TopicHandler[]
constructor(mqttBrokerUrl? : string, mqttUser? : string, mqttPass? : string, mqttCAFile? : string) {
this._mqttBrokerUrl = (mqttBrokerUrl) ? mqttBrokerUrl : MQTT_BROKER_DEFAULT_URL
this.mqttBrokerUrl = (mqttBrokerUrl) ? mqttBrokerUrl : MQTT_BROKER_DEFAULT_URL
if (mqttUser && mqttPass) {
this._mqttOptions.username = mqttUser
this._mqttOptions.password = mqttPass
this.mqttOptions.username = mqttUser
this.mqttOptions.password = mqttPass
}
if (mqttCAFile) {
this._mqttOptions.ca = fs.readFileSync(mqttCAFile, 'ascii')
this._mqttOptions.rejectUnauthorized = true
this.mqttOptions.ca = fs.readFileSync(mqttCAFile, 'ascii')
this.mqttOptions.rejectUnauthorized = true
}
this._topicHandlers = []
this.topicHandlers = []
}
register(topic: string, label: string,
@ -56,7 +56,7 @@ export class MqttDispatcher implements IDispatcher {
newChainItem = myNewChainItem
}
let done: boolean = false
for (let topicHandler of this._topicHandlers) {
for (let topicHandler of this.topicHandlers) {
if (topicHandler.topic === topic) {
(topicHandler.last as callchain.AChainItem).registerNext(newChainItem)
topicHandler.last = newChainItem
@ -65,28 +65,28 @@ export class MqttDispatcher implements IDispatcher {
}
}
if (! done) {
this._topicHandlers.push({topic: topic, root: newChainItem, last: newChainItem})
this.topicHandlers.push({topic: topic, root: newChainItem, last: newChainItem})
log.info(`first callback ${newChainItem.toString()} added for topic ${topic}`)
}
}
exec() : void {
for (let topicHandler of this._topicHandlers) {
for (let topicHandler of this.topicHandlers) {
(topicHandler.root as callchain.ChainItem).begin()
}
log.info(`connecting to ${this._mqttBrokerUrl}`)
this._mqttClient = Mqtt.connect(this._mqttBrokerUrl, this._mqttOptions)
this._mqttClient.on('error', log.error)
this._mqttClient.on('connect', (): void => {
log.info(`connecting to ${this.mqttBrokerUrl}`)
this.mqttClient = Mqtt.connect(this.mqttBrokerUrl, this.mqttOptions)
this.mqttClient.on('error', log.error)
this.mqttClient.on('connect', (): void => {
log.info("connected to mqtt broker")
for (let topicHandler of this._topicHandlers) {
this._mqttClient.subscribe(topicHandler.topic)
for (let topicHandler of this.topicHandlers) {
this.mqttClient.subscribe(topicHandler.topic)
}
})
this._mqttClient.on('message', (topic: string, payload: Buffer): void => {
this.mqttClient.on('message', (topic: string, payload: Buffer): void => {
log.info(`message received, topic ${topic}, payload ${payload}`)
for (let topicHandler of this._topicHandlers) {
for (let topicHandler of this.topicHandlers) {
if (this.topicMatch(topicHandler.topic, topic)) {
log.info(`received topic ${topic} matches registered topic ${topicHandler.topic}`);
(topicHandler.root as callchain.ChainItem).send(payload)

10
src/processor.ts
View File

@ -3,17 +3,17 @@ import * as events from 'events'
export abstract class AProcessor extends events.EventEmitter {
protected _label : string
protected label : string
constructor(label : string) {
super()
this._label = label
this.label = label
this.addListener('input', this.process)
log.info(`Processor object instanciated: ${this.constructor.name}, ${this._label}`)
log.info(`Processor object instanciated: ${this.constructor.name}, ${this.label}`)
}
public in(message : any) : void {
log.info(`Routing ${message} to Processor class ${this.constructor.name}, ${this._label}`)
log.info(`Routing ${message} to Processor class ${this.constructor.name}, ${this.label}`)
this.emit('input', message)
}
@ -23,6 +23,6 @@ export abstract class AProcessor extends events.EventEmitter {
export class ExProc1 extends AProcessor {
protected process(message : any) : void {
log.info(`ExRoute1.process: ${this._label}, ${message}`)
log.info(`ExRoute1.process: ${this.label}, ${message}`)
}
}

5
src/utils.ts
View File

@ -1,9 +1,8 @@
export function jsonPrepaper(obj:any, hideKeys:string[]) : any {
let dup = {}
for (let key in obj) {
if ((hideKeys.indexOf(key) == -1) && ! ((key[0] == "_") && (key[1] == "_"))) {
let dkey = (key[0] == "_") ? key.slice(1) : key
dup[dkey] = obj[key]
if ((hideKeys.indexOf(key) == -1) && ! (key[0] == "_")) {
dup[key] = obj[key]
}
}
return dup