fix pipeline

src/bus/can.ts

@@ -24,7 +24,7 @@ export class CANBus extends Bus {
         std.log('starting can bus', {options: this.options})
         await actions.vcan.start({name: this.options.name})
 
-        this.channel = can.createRawChannel(this.options.name)
+        this.channel = can.createRawChannelWithOptions(this.options.name, {non_block_send: true})
         this.channel.addListener('onMessage', function (message: CANMessage) {
             std.log('can bus received', {message: Message.fromCAN(message).toJSON()})
         })

src/target/can.ts

@@ -23,7 +23,7 @@ export class CANTarget extends Target {
 
     async start() {
         std.log('starting can target', {options: this.options})
-        this.target = can.createRawChannel(this.options.name)
+        this.target = can.createRawChannelWithOptions(this.options.name, {non_block_send: true})
         // TODO: does this have a site-effect on the os?
         this.target.start()
 
@@ -50,7 +50,11 @@ export class CANTarget extends Target {
         std.log('stopping can target')
         await hae.try(async () => {
             if (check.isUndefined(this.target)) return std.log('can target undefined')
-            this.target.stop()
+            try {
+                this.target.stop()
+            } catch (error) {
+                if (!error.message.includes('Channel not started')) throw error
+            }
         }, 'problem when stopping can target')
         std.log('can target stopped')
     }

src/target/file.ts

@@ -1,14 +1,16 @@
 import Target from '#/target/target'
+import * as check from '#check'
 import Message from '#core/message'
-import * as files from '#files'
 import std from '#std'
+import fs from 'node:fs'
 
 export type FileTargetOptions = {
     file: string
 }
 
 export class FileTarget extends Target {
     options: FileTargetOptions
+    stream?: fs.WriteStream
 
     constructor(options: FileTargetOptions) {
         super()
@@ -17,19 +19,39 @@ export class FileTarget extends Target {
 
     async start() {
         std.log('starting file target')
+
+        this.stream = fs.createWriteStream(this.options.file, {flags: 'a'})
         this.setReady()
+
         std.log('file target started')
     }
 
     async stop() {
         std.log('stopping file target')
+
+        await new Promise<void>((resolve, reject) => {
+            if (check.isUndefined(this.stream)) {
+                std.log('file not opened')
+                return resolve()
+            }
+            this.stream.end(() => resolve())
+        })
+
         std.log('file target stopped')
     }
 
     async send(message: Message) {
         std.log('file target sending', {message})
         message.clean()
-        await files.appendFile(this.options.file, message.toString() + '\n')
+
+        await new Promise<void>((resolve, reject) => {
+            if (check.isUndefined(this.stream)) return std.log('file not opened')
+            this.stream.write(message.toString() + '\n', error => {
+                if (check.isDefined(error)) std.log(error)
+                return resolve()
+            })
+        })
+
         std.log('file target sent')
     }
 }

src/utils/files.ts

@@ -71,15 +71,6 @@ export async function createFile(file: string) {
     await handle.close()
 }
 
-export async function appendFile(file: string, data: string) {
-    return new Promise((resolve, reject) => {
-        fs.appendFile(path.resolve(file), data, error => {
-            if (check.isDefined(error)) return reject(error)
-            return resolve(file)
-        })
-    })
-}
-
 export function storeFile(file: string, data: string) {
     fs.writeFileSync(path.resolve(file), data)
     return file

tests/bridge/file.test.ts

@@ -39,7 +39,7 @@ describe('bridge', () => {
         })
 
         std.log('waiting for message being bridged')
-        await utils.sleep(25)
+        await utils.sleep(500)
 
         expect(files.loadFile(output).trim().split('\n')).to.deep.equal([message.toString(), message.toString()])
 

tests/bridge/utils.ts

@@ -23,7 +23,9 @@ export function createBridgeTest(
 
         it(name, async () => {
             const message = Message.fromJSON({id: 69, data: [1, 2, 3], ext: false, rtr: false})
+
             const output = files.temporary()
+            await files.createFile(output)
 
             // Start can source with file target
             const bridge = await actions.bridge.start({
@@ -43,7 +45,7 @@ export function createBridgeTest(
             })
 
             std.log('waiting for message being bridged')
-            await utils.sleep(25)
+            await utils.sleep(500)
 
             const result = files.loadFile(output).trim()
             const expected = message.toString()
@@ -86,13 +88,17 @@ export function createBidirectionalBridgeTest(
         it(name, async () => {
             const request = Message.fromJSON({id: 69, data: [1, 2, 3], ext: false, rtr: false})
             const response = Message.fromJSON({id: 42, data: [4, 5, 6], ext: false, rtr: false})
+
             const output = files.temporary()
+            await files.createFile(output)
 
             // Receives "request" and returns "answer"
-            const receiver = can.createRawChannel(cans[1])
-            receiver.addListener('onMessage', function (message: CANMessage) {
+            const receiver = can.createRawChannelWithOptions(cans[1], {non_block_send: true})
+            receiver.addListener('onMessage', async function (message: CANMessage) {
                 std.log('receiver received', {message})
 
+                await utils.sleep(250)
+
                 std.log('ensuring that received message is request')
                 expect(Message.fromCAN(message).toString(), 'received message is not request').to.equal(
                     request.toString()

tests/bus/utils.ts

@@ -29,6 +29,7 @@ export function createBusTest(name: string, busOptions: BusOptions, bridgeTarget
 
             // Client 1
             const file1 = files.temporary()
+            await files.createFile(file1)
             const logger1 = await actions.bridge.start({
                 source: 'can',
                 sourceName: cans[0],
@@ -43,6 +44,7 @@ export function createBusTest(name: string, busOptions: BusOptions, bridgeTarget
 
             // Client 2
             const file2 = files.temporary()
+            await files.createFile(file2)
             const logger2 = await actions.bridge.start({
                 source: 'can',
                 sourceName: cans[1],
@@ -57,6 +59,7 @@ export function createBusTest(name: string, busOptions: BusOptions, bridgeTarget
 
             // Client 3
             const file3 = files.temporary()
+            await files.createFile(file3)
             const logger3 = await actions.bridge.start({
                 source: 'can',
                 sourceName: cans[2],
@@ -81,7 +84,7 @@ export function createBusTest(name: string, busOptions: BusOptions, bridgeTarget
             })
 
             std.log('waiting for message being bridged')
-            await utils.sleep(250)
+            await utils.sleep(500)
 
             std.log({expected: message.toString()})
 

tests/complex/complex.test.ts

@@ -57,7 +57,7 @@ describe('complex', () => {
         })
 
         std.log('waiting for message being bridged')
-        await utils.sleep(250)
+        await utils.sleep(500)
 
         expect(files.loadFile(output).trim()).to.equal(message.toString())