Code cleanup and add test

Author: albe

src/Consumer.js

@@ -46,12 +46,21 @@ class Consumer extends stream.Readable {
         if (!fs.existsSync(consumerDirectory)) {
             mkdirpSync(consumerDirectory);
         } else {
-            // Clean up left over from failed writes
-            const files = fs.readdirSync(consumerDirectory);
-            for (let file of files) {
-                if (file.startsWith(this.fileName + '.')) {
-                    fs.unlinkSync(file);
-                }
+            this.cleanUpFailedWrites(consumerDirectory);
+        }
+    }
+
+    /**
+     * Iterate over all files in the directory of this consumer and unlink any file that starts with the filename followed by a dot.
+     * @private
+     */
+    cleanUpFailedWrites() {
+        const consumerNamePrefix = path.basename(this.fileName) + '.';
+        const consumerDirectory = path.dirname(this.fileName);
+        const files = fs.readdirSync(consumerDirectory);
+        for (let file of files) {
+            if (file.startsWith(consumerNamePrefix)) {
+                fs.unlinkSync(path.join(consumerDirectory, file));
             }
         }
     }

test/Consumer.spec.js

@@ -41,6 +41,14 @@ describe('Consumer', function() {
         expect(fs.existsSync(dataDirectory + '/consumers')).to.be(true);
     });
 
+    it('cleans up failed write left-overs', function() {
+        consumer = new Consumer(storage, 'foobar', 'consumer1');
+        consumer.stop();
+        fs.writeFileSync(consumer.fileName + '.1', 'failed write!');
+        consumer = new Consumer(storage, 'foobar', 'consumer1');
+        expect(fs.existsSync(consumer.fileName + '.1')).to.be(false);
+    });
+
     it('emits event when catching up', function(done){
         consumer = new Consumer(storage, 'foobar', 'consumer1');
         consumer.stop();