CHANGELOG.md

 
+<a name="v3.3.0"></a>
+## [v3.3.0] - 2023-01-07
+### Add Features
+- change to webconnect implementation
+- hide implementation in symbol
+- new Class
+- new Webconnect and Message
+- new Webconnect and Message
+
+### Changes
+- doc
+
+
 <a name="v3.2.0"></a>
 ## [v3.2.0] - 2023-01-06
 ### Add Features
@@ -149,6 +162,7 @@
 <a name="1.8.0"></a>
 ## 1.8.0 - 2021-08-15
 
+[v3.3.0]: https://gitlab.schukai.com/oss/libraries/javascript/monster/compare/v3.2.0...v3.3.0
 [v3.2.0]: https://gitlab.schukai.com/oss/libraries/javascript/monster/compare/v3.1.2...v3.2.0
 [v3.1.2]: https://gitlab.schukai.com/oss/libraries/javascript/monster/compare/v3.1.1...v3.1.2
 [v3.1.1]: https://gitlab.schukai.com/oss/libraries/javascript/monster/compare/v3.1.0...v3.1.1

application/package.json

 {
   "name": "@schukai/monster",
-  "version": "3.1.2",
+  "version": "3.2.0",
   "description": "Monster is a simple library for creating fast, robust and lightweight websites.",
   "keywords": [
     "framework",

application/source/data/datasource/websocket.mjs

@@ -6,20 +6,15 @@
  */
 
 import {internalSymbol, instanceSymbol} from "../../constants.mjs";
-import {isInteger, isString, isObject} from "../../types/is.mjs";
-import {Queue} from "../../types/queue.mjs";
+import {isString, isObject} from "../../types/is.mjs";
+import {WebConnect} from "../../net/webconnect.mjs";
+import {Message} from "../../net/webconnect/message.mjs";
 import {Datasource} from "../datasource.mjs";
 import {Pathfinder} from "../pathfinder.mjs";
 import {Pipe} from "../pipe.mjs";
 
 export {WebSocketDatasource}
 
-/**
- * @private
- * @type {Symbol}
- */
-const receiveQueueSymbol = Symbol("queue");
-
 
 /**
  * @private
@@ -27,122 +22,33 @@ const receiveQueueSymbol = Symbol("queue");
  *
  * hint: this name is used in the tests. if you want to change it, please change it in the tests as well.
  */
-const connectionSymbol = Symbol("connection");
-
-/**
- * @private
- * @type {symbol}
- */
-const manualCloseSymbol = Symbol("manualClose");
-
-/**
- * @see https://www.rfc-editor.org/rfc/rfc6455.html#section-7.4.1
- * @type {{"1000": string, "1011": string, "1010": string, "1008": string, "1007": string, "1006": string, "1005": string, "1004": string, "1015": string, "1003": string, "1002": string, "1001": string, "1009": string}}
- */
-const connectionStatusCode = {
-    1000: "Normal closure",
-    1001: "Going away",
-    1002: "Protocol error",
-    1003: "Unsupported data",
-    1004: "Reserved",
-    1005: "No status code",
-    1006: "Connection closed abnormally",
-    1007: "Invalid frame payload data",
-    1008: "Policy violation",
-    1009: "Message too big",
-    1010: "Mandatory extension",
-    1011: "Internal server error",
-    1015: "TLS handshake"
-};
+const webConnectSymbol = Symbol("connection");
 
 /**
- * @private
- * @this {WebSocketDatasource}
- * @throws {Error} No url defined for websocket datasource.
+ *
+ * @param self
+ * @param obj
+ * @returns {*}
  */
-function connectServer(resolve, reject) {
+function doTransform(type, obj) {
     const self = this;
+    let transformation = self.getOption(type + '.mapping.transformer');
+    if (transformation !== undefined) {
+        const pipe = new Pipe(transformation);
+        const callbacks = self.getOption(type + '.mapping.callbacks')
 
-    let promiseAllredyResolved = false;
-    let connectionTimeout = self.getOption('connection.timeout');
-    if (!isInteger(connectionTimeout) || connectionTimeout < 100) {
-        connectionTimeout = 5000;
-    }
-
-    setTimeout(() => {
-        if (promiseAllredyResolved) {
-            return;
-        }
-        reject(new Error("Connection timeout"));
-    }, connectionTimeout);
-
-    let reconnectTimeout = self.getOption('connection.reconnect.timeout');
-    if (!isInteger(reconnectTimeout) || reconnectTimeout < 1000) reconnectTimeout = 1000;
-    let reconnectAttempts = self.getOption('connection.reconnect.attempts');
-    if (!isInteger(reconnectAttempts) || reconnectAttempts < 1) reconnectAttempts = 1;
-    let reconnectEnabled = self.getOption('connection.reconnect.enabled');
-    if (reconnectEnabled !== true) reconnectEnabled = false;
-
-    self[manualCloseSymbol] = false;
-    self[connectionSymbol].reconnectCounter++;
-
-    if (self[connectionSymbol].socket && self[connectionSymbol].socket.readyState < 2) {
-        self[connectionSymbol].socket.close();
-    }
-    self[connectionSymbol].socket = null;
-
-    const url = self.getOption('url');
-    if (!url) {
-        reject('No url defined for websocket datasource.');
-        return;
-    }
-
-    self[connectionSymbol].socket = new WebSocket(url);
-
-    self[connectionSymbol].socket.onmessage = function (event) {
-        self[receiveQueueSymbol].add(event);
-        setTimeout(function () {
-            self.read();
-        }, 1);
-    };
-
-    self[connectionSymbol].socket.onopen = function () {
-        self[connectionSymbol].reconnectCounter = 0;
-        if (typeof resolve === 'function' && !promiseAllredyResolved) {
-            promiseAllredyResolved = true;
-            resolve();
+        if (isObject(callbacks)) {
+            for (const key in callbacks) {
+                if (callbacks.hasOwnProperty(key) && typeof callbacks[key] === 'function') {
+                    pipe.setCallback(key, callbacks[key]);
                 }
-    };
-
-    self[connectionSymbol].socket.close = function (event) {
-
-        if (self[manualCloseSymbol]) {
-            self[manualCloseSymbol] = false;
-            return;
             }
-
-        if (reconnectEnabled && this[connectionSymbol].reconnectCounter < reconnectAttempts) {
-            setTimeout(() => {
-                self.connect();
-            }, reconnectTimeout * this[connectionSymbol].reconnectCounter);
         }
 
-    };
-
-    self[connectionSymbol].socket.onerror = (error) => {
-
-        if (reconnectEnabled && self[connectionSymbol].reconnectCounter < reconnectAttempts) {
-            setTimeout(() => {
-                self.connect();
-            }, reconnectTimeout * this[connectionSymbol].reconnectCounter);
-        } else {
-            if (typeof reject === 'function' && !promiseAllredyResolved) {
-                promiseAllredyResolved = true;
-                reject(error);
-            }
+        obj = pipe.run(obj);
     }
 
-    };
+    return obj;
 }
 
 /**
@@ -164,18 +70,25 @@ class WebSocketDatasource extends Datasource {
     constructor(options) {
         super();
         
+        const self = this;
+
         if (isString(options)) {
             options = {url: options};
         }
 
         if (!isObject(options)) options = {};
         this.setOptions(options);
-        this[receiveQueueSymbol] = new Queue();
-
-        this[connectionSymbol] = {};
-        this[connectionSymbol].socket = null;
-        this[connectionSymbol].reconnectCounter = 0;
-        this[manualCloseSymbol] = false;
+        this[webConnectSymbol] = new WebConnect({
+            url: self.getOption('url'),
+            connection: {
+                timeout: self.getOption('connection.timeout'),
+                reconnect: {
+                    timeout: self.getOption('connection.reconnect.timeout'),
+                    attempts: self.getOption('connection.reconnect.attempts'),
+                    enabled: self.getOption('connection.reconnect.enabled')
+                }
+            }
+        });
     }
 
     /**
@@ -183,18 +96,14 @@ class WebSocketDatasource extends Datasource {
      * @returns {Promise}
      */
     connect() {
-        const self = this;
-
-        return new Promise((resolve, reject) => {
-            connectServer.call(this, resolve, reject);
-        });
+        return this[webConnectSymbol].connect();
     }
 
     /**
      * @returns {boolean}
      */
     isConnected() {
-      return this[connectionSymbol]?.socket?.readyState === 1;
+        return this[webConnectSymbol].isConnected();
     }
 
     /**
@@ -216,12 +125,11 @@ class WebSocketDatasource extends Datasource {
      * @property {Object} write.mapping the mapping is applied before writing.
      * @property {String} write.mapping.transformer Transformer to select the appropriate entries
      * @property {Monster.Data.Datasource~exampleCallback[]} write.mapping.callback with the help of the callback, the structures can be adjusted before writing.
-     * @property {Object} write.report
-     * @property {String} write.report.path Path to validations
      * @property {Object} write.sheathing
      * @property {Object} write.sheathing.object Object to be wrapped
      * @property {string} write.sheathing.path Path to the data
      * @property {Object} read={} Options
+     * @property {String} read.path Path to data
      * @property {Object} read.mapping the mapping is applied after reading.
      * @property {String} read.mapping.transformer Transformer to select the appropriate entries
      * @property {Monster.Data.Datasource~exampleCallback[]} read.mapping.callback with the help of the callback, the structures can be adjusted after reading.
@@ -232,10 +140,7 @@ class WebSocketDatasource extends Datasource {
             write: {
                 mapping: {
                     transformer: undefined,
-                    callbacks: []
-                },
-                report: {
-                    path: undefined
+                    callbacks: {}
                 },
                 sheathing: {
                     object: undefined,
@@ -245,8 +150,9 @@ class WebSocketDatasource extends Datasource {
             read: {
                 mapping: {
                     transformer: undefined,
-                    callbacks: []
+                    callbacks: {}
                 },
+                path: undefined,
             },
             connection: {
                 timeout: 5000,
@@ -265,11 +171,7 @@ class WebSocketDatasource extends Datasource {
      * @returns {Promise}
      */
     close() {
-        this[manualCloseSymbol] = true;
-        if (this[connectionSymbol].socket) {
-            this[connectionSymbol].socket.close();
-        }
-        return this;
+        return this[webConnectSymbol].close();
     }
 
     /**
@@ -277,99 +179,130 @@ class WebSocketDatasource extends Datasource {
      */
     read() {
         const self = this;
-        let response;
-
-        if (self[connectionSymbol]?.socket?.readyState !== 1) {
-            return Promise.reject('The connection is not established.');
-        }
 
         return new Promise((resolve, reject) => {
-            if (self[receiveQueueSymbol].isEmpty()) {
-                resolve();
-            }
-
-            while (!self[receiveQueueSymbol].isEmpty()) {
-
-                const event = self[receiveQueueSymbol].poll();
-                const body = event?.data;
-                if (!body) continue;
-
-                let obj;
-                try {
-                    obj = JSON.parse(body);
-                } catch (e) {
 
-                    let msg = 'the response does not contain a valid json (actual: ';
-
-                    if (body.length > 100) {
-                        msg += body.substring(0, 97) + '...';
-                    } else {
-                        msg += body;
+            while (this[webConnectSymbol].dataReceived() === true) {
+                let obj = this[webConnectSymbol].poll();
+                if (!isObject(obj)) {
+                    reject(new Error('The received data is not an object.'));
+                    return;
                 }
 
-                    msg += "; " + e.message + ')';
-
-                    reject(msg);
+                if (!(obj instanceof Message)) {
+                    reject(new Error('The received data is not a Message.'));
+                    return;
                 }
 
-                let transformation = self.getOption('read.mapping.transformer');
-                if (transformation !== undefined) {
-                    const pipe = new Pipe(transformation);
+                obj = obj.getData();
 
-                    for (const callback of self.getOption('read.mapping.callbacks')) {
-                        pipe.setCallback(callback.constructor.name, callback);
+                obj = self.transformServerPayload.call(self, obj);
+                self.set( obj);
             }
 
-                    obj = pipe.run(obj);
-                }
+            resolve(self.get());
 
-                self.set(obj);
-                return response;
-            }
         })
-    }
+
+    };
+
+    // const self = this;
+    // let response;
+    //
+    // if (self[webConnectSymbol]?.socket?.readyState !== 1) {
+    //     return Promise.reject('The connection is not established.');
+    // }
+    //
+    // return new Promise((resolve, reject) => {
+    //     if (self[receiveQueueSymbol].isEmpty()) {
+    //         resolve();
+    //     }
+    //
+    //     while (!self[receiveQueueSymbol].isEmpty()) {
+    //
+    //         const event = self[receiveQueueSymbol].poll();
+    //         const body = event?.data;
+    //         if (!body) continue;
+    //
+    //         let obj;
+    //         try {
+    //             obj = JSON.parse(body);
+    //         } catch (e) {
+    //
+    //             let msg = 'the response does not contain a valid json (actual: ';
+    //
+    //             if (body.length > 100) {
+    //                 msg += body.substring(0, 97) + '...';
+    //             } else {
+    //                 msg += body;
+    //             }
+    //
+    //             msg += "; " + e.message + ')';
+    //
+    //             reject(msg);
+    //             return;
+    //         }
+    //
+    //         obj = self.transformServerPayload.call(self, obj);
+    //
+    //
+    //         self.set(obj);
+    //         return response;
+    //     }
+    // })
+//}
 
     /**
-     * @return {Promise}
+     * This prepares the data that comes from the server.
+     * Should not be called directly.
+     *
+     * @private
+     * @param {Object} payload
+     * @returns {Object}
      */
-    write() {
+    transformServerPayload(payload) {
         const self = this;
+        payload = doTransform.call(self, 'read', payload);
 
-        if (self[connectionSymbol]?.socket?.readyState !== 1) {
-            return Promise.reject('The connection is not established.');
+        const dataPath = self.getOption('read.path');
+        if (dataPath) {
+            payload = (new Pathfinder(payload)).getVia(dataPath);
         }
 
-        let obj = self.get();
-        let transformation = self.getOption('write.mapping.transformer');
-        if (transformation !== undefined) {
-            const pipe = new Pipe(transformation);
-
-            for (const callback of self.getOption('write.mapping.callbacks')) {
-                pipe.setCallback(callback.constructor.name, callback);
+        return payload;
     }
 
-            obj = pipe.run(obj);
-        }
+    /**
+     * This prepares the data for writing and should not be called directly.
+     *
+     * @private
+     * @param {Object} payload
+     * @returns {Object}
+     */
+    prepareServerPayload(payload) {
+        const self = this;
+
+        payload = doTransform.call(self, 'write', payload);
 
         let sheathingObject = self.getOption('write.sheathing.object');
         let sheathingPath = self.getOption('write.sheathing.path');
-        let reportPath = self.getOption('write.report.path');
 
         if (sheathingObject && sheathingPath) {
-            const sub = obj;
-            obj = sheathingObject;
-            (new Pathfinder(obj)).setVia(sheathingPath, sub);
+            const sub = payload;
+            payload = sheathingObject;
+            (new Pathfinder(payload)).setVia(sheathingPath, sub);
         }
 
-        return new Promise((resolve, reject) => {
-
-            if (self[connectionSymbol].socket.readyState !== 1) {
-                reject('the socket is not ready');
+        return payload;
     }
 
-            self[connectionSymbol].socket.send(JSON.stringify(obj))
-            resolve();
-        });
+    /**
+     * @return {Promise}
+     */
+    write() {
+        const self = this;
+        let obj = self.prepareServerPayload(self.get());
+        return self[webConnectSymbol].send(obj)
     }
 
 

application/source/net/webconnect.mjs

+/**
+ * Copyright schukai GmbH and contributors 2022. All Rights Reserved.
+ * Node module: @schukai/monster
+ * This file is licensed under the AGPLv3 License.
+ * License text available at https://www.gnu.org/licenses/agpl-3.0.en.html
+ */
+
+import {instanceSymbol} from "../constants.mjs";
+import {isInteger, isString, isObject} from "../types/is.mjs";
+import {BaseWithOptions} from "../types/basewithoptions.mjs";
+import {ObservableQueue} from "../types/observablequeue.mjs";
+import {Message} from "./webconnect/message.mjs";
+
+
+export {WebConnect}
+
+/**
+ * @private
+ * @type {Symbol}
+ */
+const receiveQueueSymbol = Symbol("receiveQueue");
+/**
+ * @private
+ * @type {Symbol}
+ */
+const sendQueueSymbol = Symbol("sendQueue");
+
+/**
+ * @private
+ * @type {Symbol}
+ *
+ * hint: this name is used in the tests. if you want to change it, please change it in the tests as well.
+ */
+const connectionSymbol = Symbol("connection");
+
+/**
+ * @private
+ * @type {symbol}
+ */
+const manualCloseSymbol = Symbol("manualClose");
+
+/**
+ * @private
+ * @see https://www.rfc-editor.org/rfc/rfc6455.html#section-7.4.1
+ * @type {{"1000": string, "1011": string, "1010": string, "1008": string, "1007": string, "1006": string, "1005": string, "1004": string, "1015": string, "1003": string, "1002": string, "1001": string, "1009": string}}
+ */
+const connectionStatusCode = {
+    1000: "Normal closure",
+    1001: "Going away",
+    1002: "Protocol error",
+    1003: "Unsupported data",
+    1004: "Reserved",
+    1005: "No status code",
+    1006: "Connection closed abnormally",
+    1007: "Invalid frame payload data",
+    1008: "Policy violation",
+    1009: "Message too big",
+    1010: "Mandatory extension",
+    1011: "Internal server error",
+    1015: "TLS handshake"
+};
+
+/**
+ * @private
+ * @this {WebConnect}
+ * @throws {Error} No url defined for websocket datasource.
+ */
+function connectServer(resolve, reject) {
+    const self = this;
+
+    const url = self.getOption('url');
+    if (!url) {
+        reject('No url defined for webconnect.');
+        return;
+    }
+
+    let promiseAllredyResolved = false;
+
+    let connectionTimeout = self.getOption('connection.timeout');
+    if (!isInteger(connectionTimeout) || connectionTimeout < 100) {
+        connectionTimeout = 5000;
+    }
+
+    setTimeout(() => {
+        if (promiseAllredyResolved) {
+            return;
+        }
+        reject(new Error("Connection timeout"));
+    }, connectionTimeout);
+
+    let reconnectTimeout = self.getOption('connection.reconnect.timeout');
+    if (!isInteger(reconnectTimeout) || reconnectTimeout < 1000) reconnectTimeout = 1000;
+    let reconnectAttempts = self.getOption('connection.reconnect.attempts');
+    if (!isInteger(reconnectAttempts) || reconnectAttempts < 1) reconnectAttempts = 1;
+    let reconnectEnabled = self.getOption('connection.reconnect.enabled');
+    if (reconnectEnabled !== true) reconnectEnabled = false;
+
+    self[manualCloseSymbol] = false;
+    self[connectionSymbol].reconnectCounter++;
+
+    if (self[connectionSymbol].socket && self[connectionSymbol].socket.readyState < 2) {
+        self[connectionSymbol].socket.close();
+    }
+    self[connectionSymbol].socket = null;
+    self[connectionSymbol].socket = new WebSocket(url);
+
+    self[connectionSymbol].socket.onmessage = function (event) {
+        if (event.data instanceof Blob) {
+            const reader = new FileReader();
+            reader.addEventListener("loadend", function () {
+                self[receiveQueueSymbol].add(new Message(reader.result))
+            });
+            reader.readAsText(new Message(event.data));
+        } else {
+            self[receiveQueueSymbol].add(Message.fromJSON(event.data));
+        }
+    };
+
+    self[connectionSymbol].socket.onopen = function () {
+        self[connectionSymbol].reconnectCounter = 0;
+        if (typeof resolve === 'function' && !promiseAllredyResolved) {
+            promiseAllredyResolved = true;
+            resolve();
+        }
+    };
+
+    self[connectionSymbol].socket.close = function (event) {
+
+        if (self[manualCloseSymbol]) {
+            self[manualCloseSymbol] = false;
+            return;
+        }
+
+        if (reconnectEnabled && this[connectionSymbol].reconnectCounter < reconnectAttempts) {
+            setTimeout(() => {
+                self.connect();
+            }, reconnectTimeout * this[connectionSymbol].reconnectCounter);
+        }
+
+    };
+
+    self[connectionSymbol].socket.onerror = (error) => {
+
+        if (reconnectEnabled && self[connectionSymbol].reconnectCounter < reconnectAttempts) {
+            setTimeout(() => {
+                self.connect();
+            }, reconnectTimeout * this[connectionSymbol].reconnectCounter);
+        } else {
+            if (typeof reject === 'function' && !promiseAllredyResolved) {
+                promiseAllredyResolved = true;
+                reject(error);
+            }
+        }
+
+    };
+}
+
+/**
+ * The RestAPI is a class that enables a REST API server.
+ *
+ * @externalExample ../../../example/data/storage/restapi.mjs
+ * @license AGPLv3
+ * @since 3.1.0
+ * @copyright schukai GmbH
+ * @memberOf Monster.Data.Datasource
+ * @summary The LocalStorage class encapsulates the access to data objects.
+ */
+class WebConnect extends BaseWithOptions {
+
+    /**
+     *
+     * @param {Object} [options] options contains definitions for the datasource.
+     */
+    constructor(options) {
+
+        if (isString(options)) {
+            options = {url: options};
+        }
+        
+        super(options);
+
+        this[receiveQueueSymbol] = new ObservableQueue();
+        this[sendQueueSymbol] = new ObservableQueue();
+
+        this[connectionSymbol] = {};
+        this[connectionSymbol].socket = null;
+        this[connectionSymbol].reconnectCounter = 0;
+        this[manualCloseSymbol] = false;
+    }
+
+    /**
+     *
+     * @returns {Promise}
+     */
+    connect() {
+        const self = this;
+
+        return new Promise((resolve, reject) => {
+            connectServer.call(this, resolve, reject);
+        });
+    }
+
+    /**
+     * @returns {boolean}
+     */
+    isConnected() {
+        return this[connectionSymbol]?.socket?.readyState === 1;
+    }
+
+    /**
+     * This method is called by the `instanceof` operator.
+     * @returns {symbol}
+     */
+    static get [instanceSymbol]() {
+        return Symbol.for("@schukai/monster/net/webconnect");
+    }
+
+    /**
+     * @property {string} url=undefined Defines the resource that you wish to fetch.
+     * @property {Object} connection
+     * @property {Object} connection.timeout=5000 Defines the timeout for the connection.
+     * @property {Number} connection.reconnect.timeout The timeout in milliseconds for the reconnect.
+     * @property {Number} connection.reconnect.attempts The maximum number of reconnects.
+     * @property {Bool} connection.reconnect.enabled If the reconnect is enabled.
+     */
+    get defaults() {
+        return Object.assign({}, super.defaults, {
+            url: undefined,
+            connection: {
+                timeout: 5000,
+                reconnect: {
+                    timeout: 1000,
+                    attempts: 1,
+                    enabled: false,
+                }
+            }
+        });
+    }
+
+    /**
+     * This method closes the connection.
+     *
+     * @param {Number} [code=1000] The close code.
+     * @param {String} [reason=""] The close reason.
+     * @returns {Promise}
+     * @see https://www.rfc-editor.org/rfc/rfc6455.html#section-7.4.1
+     */
+    close(statusCode, reason ) {
+        if (!isInteger(statusCode) || statusCode < 1000 || statusCode > 4999) {
+            statusCode = 1000;
+        }
+        if (!isString(reason)) {
+            reason = '';
+        }
+
+        return new Promise((resolve, reject) => {
+            try {
+                this[manualCloseSymbol] = true;
+                if (this[connectionSymbol].socket) {
+                    this[connectionSymbol].socket.close(statusCode, reason);
+                }
+            } catch (error) {
+                reject(error);
+            }
+            resolve();
+        });
+
+    }
+
+    /**
+     * Polls the receive queue for new messages.
+     * 
+     * @returns {Message}
+     */
+    poll() {
+        return this[receiveQueueSymbol].poll();
+    }
+
+    /**
+     * Are there any messages in the receive queue? 
+     * 
+     * @returns {boolean}
+     */
+    dataReceived() {
+        return !this[receiveQueueSymbol].isEmpty();
+    }
+
+    /**
+     * Get Message from the receive queue, but do not remove it.
+     * 
+     * @returns {Object}
+     */
+    peek() {
+        return this[receiveQueueSymbol].peek();
+    }
+
+    /**
+     * Attach a new observer
+     *
+     * @param {Observer} observer
+     * @returns {ProxyObserver}
+     */
+    attachObserver(observer) {
+        this[receiveQueueSymbol].attachObserver(observer);
+        return this;
+    }
+
+    /**
+     * Detach a observer
+     *
+     * @param {Observer} observer
+     * @returns {ProxyObserver}
+     */
+    detachObserver(observer) {
+        this[receiveQueueSymbol].detachObserver(observer);
+        return this;
+    }
+
+    /**
+     * @param {Observer} observer
+     * @returns {boolean}
+     */
+    containsObserver(observer) {
+        return this[receiveQueueSymbol].containsObserver(observer);
+    }
+
+    /**
+     * @param {Message|Object} message
+     * @return {Promise}
+     */
+    send(message) {
+        const self = this;
+
+        return new Promise((resolve, reject) => {
+
+            if (self[connectionSymbol].socket.readyState !== 1) {
+                reject('the socket is not ready');
+            }
+
+            self[connectionSymbol].socket.send(JSON.stringify(message))
+            resolve();
+        });
+    }
+
+}
+

application/source/net/webconnect/message.mjs

+/**
+ * Copyright schukai GmbH and contributors 2022. All Rights Reserved.
+ * Node module: @schukai/monster
+ * This file is licensed under the AGPLv3 License.
+ * License text available at https://www.gnu.org/licenses/agpl-3.0.en.html
+ */
+
+import {Base} from "../../types/base.mjs";
+import {validateObject, validateString} from "../../types/validate.mjs";
+
+export {Message}
+
+const dataSymbol = Symbol("@@data");
+
+/**
+ * This class represents a WebSocket message.
+ */
+class Message extends Base {
+
+    /**
+     * @param {Object} data
+     * @throws {TypeError} value is not a object
+     */
+    constructor(data) {
+        super();
+        this[dataSymbol] = validateObject(data);
+    }
+    
+    /**
+     * Returns the raw message.
+     * 
+     * @returns {object}
+     */
+    getData() {
+        return this[dataSymbol];
+    }
+
+    /**
+     * @returns {*}
+     */
+    toJSON() {
+        return this[dataSymbol];
+    }
+
+    /**
+     * @param {string} json
+     * @returns {Message}
+     * @throws {TypeError} value is not a string
+     */
+    static fromJSON(json) {
+        validateString(json);
+        return new Message(JSON.parse(json));
+    }
+
+}
\ No newline at end of file

application/source/types/observablequeue.mjs

+/**
+ * Copyright schukai GmbH and contributors 2022. All Rights Reserved.
+ * Node module: @schukai/monster
+ * This file is licensed under the AGPLv3 License.
+ * License text available at https://www.gnu.org/licenses/agpl-3.0.en.html
+ */
+
+import {Queue} from './queue.mjs';
+import {instanceSymbol, internalSymbol} from '../constants.mjs';
+import {ObserverList} from "./observerlist.mjs";
+
+export {ObservableQueue};
+
+/**
+ * An observable queue is a list of items that are processed one after another (first in, first out).
+ * 
+ * `Queue.add()` and `Queue.clear()` notify all observers.
+ * 
+ * @externalExample ../../example/types/queue.mjs
+ * @license AGPLv3
+ * @since 3.3.0
+ * @copyright schukai GmbH
+ * @memberOf Monster.Types
+ * @summary An observable Queue (Fifo)
+ */
+class ObservableQueue extends Queue {
+
+    /**
+     *
+     */
+    constructor() {
+        super();
+        this[internalSymbol]= {
+            observers: new ObserverList()
+        };
+    }
+
+    /**
+     * This method is called by the `instanceof` operator.
+     * @returns {symbol}
+     * @since 2.1.0
+     */
+    static get [instanceSymbol]() {
+        return Symbol.for("@schukai/monster/types/observablequeue");
+    }
+    
+    /**
+     * Add a new element to the end of the queue.
+     *
+     * @param {*} value
+     * @returns {Queue}
+     */
+    add(value) {
+        super.add(value);
+        this.notifyObservers();
+        return this;
+    }
+
+    /**
+     * remove all entries
+     *
+     * @returns {Queue}
+     */
+    clear() {
+        super.clear();
+        this.notifyObservers();
+        return this;
+    }
+
+    /**
+     * Attach a new observer
+     *
+     * @param {Observer} observer
+     * @returns {ProxyObserver}
+     */
+    attachObserver(observer) {
+        this[internalSymbol].observers.attach(observer)
+        return this;
+    }
+
+    /**
+     * Detach a observer
+     *
+     * @param {Observer} observer
+     * @returns {ProxyObserver}
+     */
+    detachObserver(observer) {
+        this[internalSymbol].observers.detach(observer)
+        return this;
+    }
+
+    /**
+     * Notify all observer
+     *
+     * @returns {Promise}
+     */
+    notifyObservers() {
+        return this[internalSymbol].observers.notify(this);
+    }
+
+    /**
+     * @param {Observer} observer
+     * @returns {boolean}
+     */
+    containsObserver(observer) {
+        return this[internalSymbol].observers.contains(observer)
+    }    
+
+
+}

application/source/types/proxyobserver.mjs

@@ -61,9 +61,9 @@ export {ProxyObserver}
 
 
     /**
-     * get the real object
+     * Get the real object
      *
-     * changes to this object are not noticed by the observers, so you can make a large number of changes and inform the observers later.
+     * Changes to this object are not noticed by the observers, so you can make a large number of changes and inform the observers later.
      *
      * @returns {object}
      */

application/source/types/uniquequeue.mjs

@@ -6,12 +6,13 @@
  */
 
 import {Queue} from "./queue.mjs";
+import {internalSymbol} from "../constants.mjs";
 import {validateObject} from "./validate.mjs";
 
 export {UniqueQueue}
 
 /**
- * A UniqueQueue is a queue that contains items only once.
+ * An UniqueQueue is a queue that contains items only once.
  *
  * @license AGPLv3
  * @since 1.4.0
@@ -26,7 +27,9 @@ export {UniqueQueue}
      */
     constructor() {
         super();
-        this.unique = new WeakSet();
+        this[internalSymbol]={
+            unique : new WeakSet()
+        };
     }
 
     /**
@@ -40,8 +43,8 @@ export {UniqueQueue}
 
         validateObject(value);
 
-        if (!this.unique.has(value)) {
-            this.unique.add(value);
+        if (!this[internalSymbol].unique.has(value)) {
+            this[internalSymbol].unique.add(value);
             super.add(value);
         }
 
@@ -55,7 +58,7 @@ export {UniqueQueue}
      */
     clear() {
         super.clear();
-        this.unique = new WeakSet;
+        this[internalSymbol].unique = new WeakSet;
         return this;
     }
 
@@ -71,7 +74,7 @@ export {UniqueQueue}
             return undefined;
         }
         let value = this.data.shift();
-        this.unique.delete(value);
+        this[internalSymbol].unique.delete(value);
         return value;
     }
 

application/source/types/version.mjs

@@ -149,7 +149,7 @@ function getMonsterVersion() {
     }
 
     /** don't touch, replaced by make with package.json version */
-    monsterVersion = new Version('3.1.2')
+    monsterVersion = new Version('3.2.0')
 
     return monsterVersion;
 

development/package.json

 {
   "name": "monster",
-  "version": "3.1.2",
+  "version": "3.2.0",
   "description": "monster",
   "repository": {
     "type": "git",

development/test/cases/data/datasource/websocket.mjs

@@ -25,11 +25,17 @@ describe('Websocket', function () {
             ds.close()
         }
 
-        // without this, the node test will hang 
+        // workaround: without this, the node test will hang 
         for (const sym of Object.getOwnPropertySymbols(ds)) {
             if (sym.toString() === 'Symbol(connection)') {
-                if(ds[sym]?.socket?.['terminate']) {
-                    ds[sym]?.socket?.['terminate']()
+                const connection = ds[sym]
+                for (const sym2 of Object.getOwnPropertySymbols(connection)) {
+                    if (sym2.toString() === 'Symbol(connection)') {
+                        const socket = connection[sym2]?.socket;
+                        if (socket) {
+                            socket.terminate()
+                        }
+                    }
                 }
             }
         }
@@ -45,6 +51,84 @@ describe('Websocket', function () {
         expect(clone).to.be.an.instanceof(WebSocketDatasource)
 
 
+    })
+
+    it('should transform data', function (done) {
+
+        let writeCallbackCalled = false
+        let readCallbackCalled = false
+
+        ds = new WebSocketDatasource({
+            url: testUrl,
+            write: {
+                mapping: {
+                    transformer: "call:onWrite",
+                    callbacks: {
+                        onWrite: (data) => {
+                            writeCallbackCalled = true
+                            return data
+                        }
+                    }
+                },
+                sheathing: {
+                    object: {
+                        demo: 1,
+                        data: {
+                            xyz: undefined
+                        }
+                    },
+                    path: "data.xyz",
+                },
+            },
+            read: {
+                mapping: {
+                    transformer: "call:onRead",
+                    callbacks: {
+                        onRead: (data) => {
+                            readCallbackCalled = true
+                            return data
+                        }
+                    }
+                },
+                path: 'data.xyz',
+            }
+        })
+
+        ds.connect().then(() => {
+            ds.set({
+                envelop: {
+                    message: "Hello World"
+                }
+            })
+
+            ds.write().then(() => {
+
+                ds.set({})
+                expect(ds.get()).to.be.deep.equal({});
+
+                setTimeout(() => {
+
+
+                    ds.read().then(() => {
+                        expect(ds.get()).to.be.deep.equal({envelop:{message: "Hello World"}});
+                        expect(writeCallbackCalled).to.be.true
+                        expect(readCallbackCalled).to.be.true
+                        done()
+                    }).catch((e) => {
+                        done(e)
+                    })
+                }, 200)
+
+            }).catch((err) => {
+                done(new Error(err));
+            })
+
+
+        }).catch((e) => {
+            done(e)
+        })
+
+
     })
 
     it('should connect', function (done) {
@@ -54,11 +138,11 @@ describe('Websocket', function () {
                 enabled: false
             }
         });
-        ds.connect()
-        setTimeout(() => {
-            expect(ds.isConnected()).to.be.true;
-            done();
-        }, 500);
+        ds.connect().then(() => {
+            done()
+        }).catch((e) => {
+            done(e)
+        })
 
 
     })
@@ -70,36 +154,37 @@ describe('Websocket', function () {
                 enabled: false
             }
         });
-        ds.connect()
-
+        ds.connect().then(() => {
             ds.set({
-            data: {
+                envelop: {
                     message: "Hello World"
                 }
             })
 
-        setTimeout(() => {
-
             ds.write().then(() => {
 
                 ds.set({})
                 expect(ds.get()).to.be.deep.equal({});
 
-
                 setTimeout(() => {
 
-                    expect(ds.get()).to.be.deep.equal({
-                        data: {
-                            message: "Hello World"
-                        }
-                    });
-                    done();
-                }, 1000);
+                    ds.read().then(() => {
+                    expect(ds.get()).to.be.deep.equal({envelop:{message: "Hello World"}});
+                    done()
+                }).catch((e) => {
+                    done(e)
+                })
+                },500)
+                
+
             }).catch((err) => {
                 done(new Error(err));
             })
-        }, 
-            500)
+
+
+        }).catch((e) => {
+            done(e)
+        })
 
 
     }).timeout(10000);

development/test/cases/monster.mjs

@@ -7,7 +7,7 @@ describe('Monster', function () {
         let monsterVersion
 
         /** don´t touch, replaced by make with package.json version */
-        monsterVersion = new Version('3.1.2')
+        monsterVersion = new Version('3.2.0')
 
         let m = getMonsterVersion();
 

development/test/cases/net/webconnect.mjs

+import {expect} from "chai"
+import {WebConnect} from "../../../../application/source/net/webconnect.mjs";
+import {Message} from "../../../../application/source/net/webconnect/message.mjs";
+import {Observer} from "../../../../application/source/types/observer.mjs";
+import {initWebSocket} from "../../util/websocket.mjs";
+
+const testUrl = "wss://ws.postman-echo.com/raw"
+
+describe('Websocket', function () {
+
+    let ds = undefined
+
+    before(function (done) {
+        initWebSocket().then(() => {
+            done()
+        }).catch((e) => {
+            done(e)
+        })
+    });
+
+    afterEach(function (done) {
+        if (ds) {
+            ds.close()
+        }
+
+        // without this, the node test will hang 
+        for (const sym of Object.getOwnPropertySymbols(ds)) {
+            if (sym.toString() === 'Symbol(connection)') {
+                if (ds[sym]?.socket?.['terminate']) {
+                    ds[sym]?.socket?.['terminate']()
+                }
+            }
+        }
+
+        done()
+    });
+
+
+    it('should transform data', function (done) {
+
+        ds = new WebConnect( {
+            url: testUrl,
+        })
+
+        ds.connect().then(() => {
+            
+            ds.attachObserver(new Observer(()=> {
+               done()
+            }))
+            
+            ds.send({
+                data: {
+                    message: "Hello World"
+                }
+            })
+
+        }).catch((e) => {
+            done(e)
+        })
+
+
+    })
+
+    it('should connect', function (done) {
+        ds = new WebConnect({
+            url: testUrl,
+            reconnect: {
+                enabled: false
+            }
+        });
+        ds.connect().then(() => {
+            done()
+        }).catch((e) => {
+            done(e)
+        })
+
+
+    })
+
+    it('should send message', function (done) {
+        ds = new WebConnect({
+            url: testUrl,
+            reconnect: {
+                enabled: false
+            }
+        });
+        ds.connect().then(() => {
+
+            ds.attachObserver(new Observer(()=> {
+                
+                expect(ds.dataReceived()).to.be.true
+                
+                try {
+                    const msg = ds.poll()
+                    expect(msg).to.be.instanceOf(Message)
+                    const data = msg.getData()
+                    expect(data).to.be.deep.equal({message: "Hello World"})
+                } catch (e) {
+                    done(e)
+                    return
+                }
+                done()
+            }))
+
+            ds.send({
+                    message: "Hello World"
+            })
+
+        }).catch((e) => {
+            done(e)
+        })
+
+
+    }).timeout(10000);
+
+});

development/test/cases/net/webconnect/message.mjs

+import {expect} from "chai"
+import {Message} from "../../../../../application/source/net/webconnect/message.mjs";
+
+describe('Message', function () {
+
+    it('construct withouth parameters should throw', function (done) {
+        
+        try {
+            new Message();
+            done(new Error('should throw'));
+        } catch (e) {
+            done();
+            return;
+        }
+
+    })
+    
+    it('from json should ' , function (done) {
+        const json = {
+            "id": "123",
+            "type": "test",
+            "data": {
+                "test": "test"
+            }
+        }
+        const message = Message.fromJSON(JSON.stringify(json));
+        const data = message.getData();
+        expect(data.id).to.equal(json.id);
+        expect(data.type).to.equal(json.type);
+        expect(data.data).to.deep.equal(json.data);
+        done();
+    })
+    
+    it ("to json should", function (done) {
+        const obj = {
+            "id": "123",
+            "type": "test",
+            "data": {
+                "test": "test"
+            }
+        }
+        const message = new Message(obj);
+        const data = JSON.stringify(message);
+        expect(data).to.equal('{"id":"123","type":"test","data":{"test":"test"}}');
+        done();
+    })
+
+
+
+});

development/test/cases/types/observablequeue.mjs

+import {expect} from "chai"
+import {ObservableQueue} from "../../../../application/source/types/observablequeue.mjs";
+import {Observer} from "../../../../application/source/types/observer.mjs";
+
+describe('ObservableQueue', function () {
+    describe('Observer', function () {
+
+        it('should notify', function (done) {
+            let queue = new ObservableQueue;
+            let o = new Observer((q) => {
+                done()
+            });
+            queue.attachObserver(o);
+            expect(queue.add('a')).to.be.instanceOf(ObservableQueue);
+        });
+    });
+})
\ No newline at end of file

development/test/cases/types/queue.mjs

@@ -42,4 +42,7 @@ describe('Queue', function () {
         });
 
     })
+
+
+    
 })
\ No newline at end of file

release.json

-{"version":"3.2.0"}
+{"version":"3.3.0"}