Send message when transfer finished #244
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -57,6 +57,7 @@ transferRouter.post( | |
| const walletLogin = await walletService.getById(res.locals.wallet_id); | ||
| const walletSender = await walletService.getByIdOrName(req.body.sender_wallet); | ||
| const walletReceiver = await walletService.getByIdOrName(req.body.receiver_wallet); | ||
| const transferService = new TransferService(session); | ||
| // check if this transfer is a claim (claim == not transferrrable tokens) | ||
| const claim = req.body.claim; | ||
|
|
||
|
|
@@ -76,7 +77,18 @@ transferRouter.post( | |
| // TODO: get only transferrable tokens | ||
| result = await walletLogin.transferBundle(walletSender, walletReceiver, req.body.bundle.bundle_size, claim); | ||
| } | ||
|
|
||
| // send message | ||
| if (result.state === Transfer.STATE.completed) { | ||
| // just send message in production | ||
| if(process.env.NODE_ENV !== "test"){ | ||
| await transferService.sendMessage(result.id); | ||
|
There was a problem hiding this comment. One way we could make this a bit more robust is to raise the messages as a domain-event (a separate table that stores all the events raised) with a payload that holds the message and a status(that can be
Collaborator
Author
There was a problem hiding this comment. Do we have some 3rd party libraries to implement this pattern? Or we should build a module about this by ourselves? I think this might be a common demand in different our project? There was a problem hiding this comment. You are right, we need this in whichever service we emit/receive messages and store/process them. I started a separate project (to be treated as a submodule) https://github.com/Greenstand/domain-events. I am not sure if there is a NPM library that can meet our needs exactly. It might easier to use our own given we also need to store these events in a database. |
||
| } | ||
| } | ||
|
|
||
| await session.commitTransaction(); | ||
|
|
||
| // response | ||
| result = await transferService.convertToResponse(result); | ||
| if (result.state === Transfer.STATE.completed) { | ||
| res.status(201).json(result); | ||
|
|
@@ -88,7 +100,6 @@ transferRouter.post( | |
| } else { | ||
| throw new Error(`Unexpected state ${result.state}`); | ||
| } | ||
| }catch(e){ | ||
| if(e instanceof HttpError && !e.shouldRollback()){ | ||
| // if the error type is HttpError, means the exception has been handled | ||
|
|
@@ -127,8 +138,12 @@ transferRouter.post( | |
| const transferJson2 = await transferService.convertToResponse( | ||
| transferJson, | ||
| ); | ||
| // just send message in production | ||
| if(process.env.NODE_ENV !== "test"){ | ||
| await transferService.sendMessage(transferJson.id); | ||
| } | ||
| await session.commitTransaction(); | ||
| res.status(200).json(transferJson2); | ||
| } catch (e) { | ||
| if (e instanceof HttpError && !e.shouldRollback()) { | ||
| // if the error type is HttpError, means the exception has been handled | ||
|
|
@@ -167,8 +182,8 @@ transferRouter.post( | |
| const transferJson2 = await transferService.convertToResponse( | ||
| transferJson, | ||
| ); | ||
| await session.commitTransaction(); | ||
| res.status(200).json(transferJson2); | ||
| } catch (e) { | ||
| if (e instanceof HttpError && !e.shouldRollback()) { | ||
| // if the error type is HttpError, means the exception has been handled | ||
|
|
@@ -207,8 +222,8 @@ transferRouter.delete( | |
| const transferJson2 = await transferService.convertToResponse( | ||
| transferJson, | ||
| ); | ||
| await session.commitTransaction(); | ||
| res.status(200).json(transferJson2); | ||
| } catch (e) { | ||
| if (e instanceof HttpError && !e.shouldRollback()) { | ||
| // if the error type is HttpError, means the exception has been handled | ||
|
|
@@ -280,8 +295,12 @@ transferRouter.post( | |
| const transferJson2 = await transferService.convertToResponse( | ||
| transferJson, | ||
| ); | ||
| // just send message in production | ||
| if(process.env.NODE_ENV !== "test"){ | ||
| await transferService.sendMessage(transferJson.id); | ||
| } | ||
| await session.commitTransaction(); | ||
| res.status(200).json(transferJson2); | ||
| } catch (e) { | ||
| if (e instanceof HttpError && !e.shouldRollback()) { | ||
| // if the error type is HttpError, means the exception has been handled | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,26 @@ | ||
| module.exports = { | ||
| config: { | ||
| "vhosts": { | ||
| "test": { | ||
| "connection": { | ||
| "url": process.env.RABBIT_MQ_URL, | ||
| "socketOptions": { | ||
| "timeout": 1000 | ||
| } | ||
| }, | ||
| "exchanges": ["wallet-service-ex"], | ||
| "queues": ["token-transfer:events"], | ||
| "bindings": [ | ||
| "wallet-service-ex[token.transfer] -> token-transfer:events" | ||
| ], | ||
| "publications": { | ||
| "token-assigned": { | ||
| "exchange": "wallet-service-ex", | ||
| "routingKey": "token.transfer" | ||
| } | ||
| }, | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,44 @@ | ||
| const Broker = require('rascal').BrokerAsPromised; | ||
| const config = require("./MQConfig").config; | ||
| const HttpError = require("../utils/HttpError"); | ||
| const log = require("loglevel"); | ||
|
|
||
| class MQService{ | ||
|
|
||
| constructor(session){ | ||
| this._settsion = session; | ||
| } | ||
|
|
||
| sendMessage(payload){ | ||
| log.warn("to send message"); | ||
| return new Promise((resolve, reject) => { | ||
| Broker.create(config) | ||
| .then(broker => { | ||
| broker.publish("token-assigned", payload) | ||
| .then(publication => { | ||
| log.warn("publication is on"); | ||
| publication | ||
| .on("success", () => { | ||
| log.warn("message sent!"); | ||
| resolve(true); | ||
| }) | ||
| .on("error", (err, messageId)=> { | ||
| const error = `Error with id ${messageId} ${err.message}`; | ||
| log.error(error); | ||
| reject(new HttpError(500, error)); | ||
| }); | ||
| }) | ||
| .catch(err => { | ||
| log.error(err); | ||
| reject(new HttpError(500, `Error publishing message ${err}`)); | ||
| }) | ||
| }) | ||
| .catch(err => { | ||
| log.error(err); | ||
| reject(new HttpError(500, `Error create broker ${err}`)); | ||
| }) | ||
| }); | ||
| } | ||
| } | ||
|
|
||
| module.exports = MQService; |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,103 @@ | ||
| const MQService = require("./MQService"); | ||
| const Broker = require('rascal').BrokerAsPromised; | ||
| const sinon = require("sinon"); | ||
| const {expect} = require("chai"); | ||
| const jestExpect = require("expect"); | ||
| const log = require("loglevel"); | ||
|
|
||
| describe("MQService", () => { | ||
|
|
||
| afterEach(() => { | ||
| sinon.restore(); | ||
| }); | ||
|
|
||
| it("send message successfully", async () => { | ||
| const broker = { | ||
| publish: async () => { | ||
| console.log("publish"); | ||
| return { | ||
| on(event, handler){ | ||
| // mock the success event | ||
| if(event === "success"){ | ||
| setImmediate(handler); | ||
| } | ||
| return this; | ||
| } | ||
| } | ||
| } | ||
| }; | ||
| sinon.spy(broker, "publish"); | ||
| sinon.stub(Broker, "create").resolves(broker); | ||
| const mqService = new MQService(); | ||
|
|
||
| const payload = {a:1}; | ||
| const result = await mqService.sendMessage(payload); | ||
| expect(result).eq(true); | ||
| sinon.assert.calledWith(broker.publish, "raw-capture-created", payload, "field-data.capture.creation"); | ||
|
|
||
| }); | ||
|
|
||
| it("send message with problem", async () => { | ||
| sinon.stub(Broker, "create").resolves({ | ||
| publish: async () => { | ||
| console.log("publish"); | ||
| return { | ||
| on(event, handler){ | ||
| // mock the error event | ||
| if(event === "error"){ | ||
| setImmediate(() => handler(new Error("Message sending wrong"), "No.1")); | ||
| } | ||
| return this; | ||
| } | ||
| } | ||
| } | ||
| }); | ||
| const mqService = new MQService(); | ||
|
|
||
| await jestExpect(async () => { | ||
| await mqService.sendMessage({a:1}); | ||
| }).rejects.toThrow(/Message sending wrong/); | ||
|
|
||
| }); | ||
|
|
||
| }); | ||
|
|
||
| describe("Real operation, just for dev", () => { | ||
|
|
||
| it("Send and receive message", async function(){ | ||
| try{ | ||
|
|
||
| const mqService = new MQService(); | ||
| const payload = {a:1}; | ||
| const result = await mqService.sendMessage(payload); | ||
| log.warn("result:", result); | ||
|
|
||
|
|
||
| // await new Promise((resolve, reject) => { | ||
| // // check the message | ||
| // // Consume a message | ||
| // const config = require("./MQConfig").config; | ||
| // Broker.create(config) | ||
| // .then(broker => { | ||
| // log.info("connected to broker"); | ||
| // broker.subscribeAll() | ||
| // .then(subscriptions => { | ||
| // subscriptions.forEach( subscription => { | ||
| // subscription.on('message', (message, content, ackOrNack) => { | ||
| // log.warn("message:", message, content); | ||
| // log.warn("message content received:", message.content && message.content.toString()); | ||
| // ackOrNack(); | ||
| // resolve(); | ||
| // }).on('error', console.error); | ||
| // }); | ||
| // }); | ||
| // }); | ||
| // const mqService = new MQService(); | ||
| // const payload = {a:1}; | ||
| // mqService.sendMessage(payload); | ||
| // }); | ||
| }catch(e){ | ||
| log.error("e:",e ); | ||
| }; | ||
| }); | ||
| }); |
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Is match here a regex kind? I feel either we should do a robust match by verifying the exact query if possible.
Even if we did this test seems to be verifying just the method signature and knex but nothing significant in our code. I would prefer to skip the unit test for this Repository class if you ask me. It would be beneficial and reduce our developer time spent on maintaining it.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Hmm, maybe you are right, I don't realize that I added some tests maybe look not so useful, yes, it might be unnessarary, I will think it over, and another reason lead to this result is that I wrote the test first before I wrote the implementation, so when I link this model to the real system, I don't need to worry too much about that it's some problematic model, it makes integrating the code easier.