박권수

fix. Mqtt unnecessary call remove

...@@ -2,12 +2,10 @@ ...@@ -2,12 +2,10 @@
2 const Bottle = require('../../models/bottle'); 2 const Bottle = require('../../models/bottle');
3 const Hub = require('../../models/hub'); 3 const Hub = require('../../models/hub');
4 const Medicine = require('../../models/medicine'); 4 const Medicine = require('../../models/medicine');
5 -const DataProcess = require('../../lib/DataProcess');
6 const Mqtt = require('../../lib/MqttModule'); 5 const Mqtt = require('../../lib/MqttModule');
7 6
8 exports.bottleConnect = async(ctx) => { 7 exports.bottleConnect = async(ctx) => {
9 const { bottleId, hubId } = ctx.request.body; 8 const { bottleId, hubId } = ctx.request.body;
10 - const topic = 'bottle/' + bottleId + '/bts';
11 9
12 const newBottle = new Bottle({ 10 const newBottle = new Bottle({
13 bottleId, 11 bottleId,
...@@ -32,13 +30,9 @@ exports.bottleConnect = async(ctx) => { ...@@ -32,13 +30,9 @@ exports.bottleConnect = async(ctx) => {
32 return; 30 return;
33 } 31 }
34 32
35 - const client = await Mqtt.mqttOn({ 33 + const client = await Mqtt.mqttOn(hosting);
36 - host : hosting.host, 34 + const topic = 'bottle/' + bottleId + '/bts';
37 - port : hosting.port, 35 + Mqtt.mqttSubscribe(client, topic);
38 - clientId : hosting.clientId
39 - });
40 -
41 - Mqtt.mqttSubscribe(client, topic, DataProcess.dataPublish);
42 36
43 await newBottle.save(); 37 await newBottle.save();
44 38
...@@ -46,7 +40,25 @@ exports.bottleConnect = async(ctx) => { ...@@ -46,7 +40,25 @@ exports.bottleConnect = async(ctx) => {
46 }; 40 };
47 41
48 exports.bottleDisconnect = async(ctx) => { 42 exports.bottleDisconnect = async(ctx) => {
49 - const { bottleId } = ctx.params; 43 + const { bottleId } = ctx.params;
44 +
45 + const bottle = await Bottle.findByBottleId(bottleId);
46 + if(!bottle) {
47 + ctx.status = 404;
48 + return;
49 + }
50 +
51 + const hub = await Hub.findByHubId(bottle.getHubId());
52 + const hosting = await hub.getHubHost();
53 +
54 + const client = await Mqtt.mqttOn(hosting);
55 + const topic = 'bottle/' + bottleId + '/bts';
56 + Mqtt.mqttUnsubscribe(client, topic);
57 +
58 + await Bottle.deleteOne({ bottleId });
59 +
60 + ctx.status = 200;
61 +
50 }; 62 };
51 63
52 exports.lookupInfo = async(ctx) => { 64 exports.lookupInfo = async(ctx) => {
......
1 //허브(Mqtt Broker)등록 및 삭제 1 //허브(Mqtt Broker)등록 및 삭제
2 const Hub = require('../../models/hub'); 2 const Hub = require('../../models/hub');
3 const Mqtt = require('../../lib/MqttModule'); 3 const Mqtt = require('../../lib/MqttModule');
4 +const DataProcess = require('../../lib/DataProcess');
4 5
5 exports.hubConnect = async (ctx) => { 6 exports.hubConnect = async (ctx) => {
6 const { hubId, host, port } = ctx.request.body; 7 const { hubId, host, port } = ctx.request.body;
7 8
9 + const isExistHub = await Hub.findByHubId(hubId);
10 + if(isExistHub) {
11 + ctx.status = 409;
12 + return;
13 + }
14 +
8 const hosting = { 15 const hosting = {
9 host, 16 host,
10 port 17 port
11 }; 18 };
12 19
13 - Mqtt.mqttOn(hosting); 20 + Mqtt.mqttOn(hosting, DataProcess.dataPublish);
14 - await Hub.findOneAndUpdate({ 21 +
15 - hubId 22 + const hub = new Hub({
16 - }, { hosting }, { 23 + hubId,
17 - upsert : true 24 + hosting
18 }); 25 });
19 26
27 + await hub.save();
28 +
20 ctx.status = 200; 29 ctx.status = 200;
21 -} 30 + ctx.body = hub;
31 +};
22 32
23 exports.hubDisconnect = async(ctx) => { 33 exports.hubDisconnect = async(ctx) => {
24 const { hubId } = ctx.params; 34 const { hubId } = ctx.params;
...@@ -35,4 +45,4 @@ exports.hubDisconnect = async(ctx) => { ...@@ -35,4 +45,4 @@ exports.hubDisconnect = async(ctx) => {
35 await Hub.deleteOne({ hubId }); 45 await Hub.deleteOne({ hubId });
36 46
37 ctx.status = 200; 47 ctx.status = 200;
38 -}
...\ No newline at end of file ...\ No newline at end of file
48 +};
...\ No newline at end of file ...\ No newline at end of file
......
1 const mqtt = require('mqtt'); 1 const mqtt = require('mqtt');
2 const clientList = []; 2 const clientList = [];
3 3
4 -exports.mqttOn = async (hosting) => { 4 +exports.mqttOn = async (hosting, func) => {
5 const filterIndex = clientList.findIndex(client => { 5 const filterIndex = clientList.findIndex(client => {
6 return (client.options.clientId === hosting.clientId 6 return (client.options.clientId === hosting.clientId
7 && client.options.host === hosting.host 7 && client.options.host === hosting.host
...@@ -11,26 +11,37 @@ exports.mqttOn = async (hosting) => { ...@@ -11,26 +11,37 @@ exports.mqttOn = async (hosting) => {
11 if(filterIndex === -1) { 11 if(filterIndex === -1) {
12 const client = mqtt.connect(hosting); 12 const client = mqtt.connect(hosting);
13 clientList.push(client); 13 clientList.push(client);
14 +
14 client.on('connect', () => { 15 client.on('connect', () => {
15 console.log('Client connected: ', client.connected); 16 console.log('Client connected: ', client.connected);
16 }); 17 });
18 +
19 + client.on('message', async (topic, message, packet) => {
20 + const result = await func(topic, message.toString());
21 + this.mqttPublishMessage(client, result);
22 + console.log('\x1b[1;32msubscribe : topic', topic, 'message : ', message.toString(), '\x1b[0m');
23 + });
17 24
18 return client; 25 return client;
19 - } else { 26 + }
20 - return clientList[filterIndex]; 27 +
21 - }; 28 + return clientList[filterIndex];
22 }; 29 };
23 30
24 -exports.mqttSubscribe = (client, topic, func) => { 31 +exports.mqttSubscribe = (client, topic) => {
25 client.subscribe(topic); 32 client.subscribe(topic);
26 - client.on('message', async (topic, message, packet) => {
27 - const result = await func(topic, message.toString());
28 - this.mqttPublishMessage(client, result);
29 - });
30 }; 33 };
31 34
32 exports.mqttPublishMessage = (client, { topic, message }) => { 35 exports.mqttPublishMessage = (client, { topic, message }) => {
33 - client.publish(topic, message, () => {}); 36 + client.publish(topic, message, () => {
37 + console.log('\x1b[1;33mpublish : topic', topic, 'message : ', message, '\x1b[0m');
38 + });
39 +};
40 +
41 +exports.mqttUnsubscribe = (client, topic) => {
42 + client.unsubscribe(topic, () => {
43 + console.log('unsubscribe', topic);
44 + });
34 }; 45 };
35 46
36 exports.mqttOff = (hosting) => { 47 exports.mqttOff = (hosting) => {
......