Adding connectionOptions as a plugin option to enable Duplex options

README.md

@@ -238,7 +238,6 @@ fastify.listen(3000, err => {
 
 `fastify-websocket` accept these options for [`ws`](https://github.com/websockets/ws/blob/master/doc/ws.md#new-websocketserveroptions-callback) :
 
-- `objectMode` - Send each chunk on its own, and do not try to pack them in a single websocket frame.
 - `host` - The hostname where to bind the server.
 - `port` - The port where to bind the server.
 - `backlog` - The maximum length of the queue of pending connections.
@@ -257,6 +256,16 @@ _**NB** The `path` option from `ws` should not be provided since the routing is
 
 _**NB** The `noServer` option from `ws` should not be provided since the point of fastify-websocket is to listen on the fastify server. If you want a custom server, you can use the `server` option, and if you want more control, you can use the `ws` library directly_
 
+You can also pass the following as `connectionOptions` for [createWebSocketStream](https://github.com/websockets/ws/blob/master/doc/ws.md#createwebsocketstreamwebsocket-options).
+
+- `allowHalfOpen` <boolean> If set to false, then the stream will automatically end the writable side when the readable side ends. Default: true.
+- `readable` <boolean> Sets whether the Duplex should be readable. Default: true.
+- `writable` <boolean> Sets whether the Duplex should be writable. Default: true.
+- `readableObjectMode` <boolean> Sets objectMode for readable side of the stream. Has no effect if objectMode is true. Default: false.
+- `readableHighWaterMark` <number> Sets highWaterMark for the readable side of the stream.
+- `writableHighWaterMark` <number> Sets highWaterMark for the writable side of the stream.
+
+[ws](https://github.com/websockets/ws) does not allow you to set `objectMode` or `writableObjectMode` to true
 ## Acknowledgements
 
 This project is kindly sponsored by [nearForm](https://nearform.com).

index.d.ts

@@ -3,7 +3,7 @@ import { IncomingMessage, ServerResponse, Server } from 'http';
 import { FastifyRequest, FastifyPluginCallback, RawServerBase, RawServerDefault, RawRequestDefaultExpression, RawReplyDefaultExpression, RequestGenericInterface, ContextConfigDefault, FastifyInstance } from 'fastify';
 import * as fastify from 'fastify';
 import * as WebSocket from 'ws';
-import { Duplex } from 'stream';
+import { Duplex, DuplexOptions } from 'stream';
 import { FastifyReply } from 'fastify/types/reply';
 import { RouteGenericInterface } from 'fastify/types/route';
 
@@ -59,6 +59,7 @@ export interface SocketStream extends Duplex {
 export interface WebsocketPluginOptions {
   errorHandler?: (this: FastifyInstance, error: Error, connection: SocketStream, request: FastifyRequest, reply: FastifyReply) => void;
   options?: WebSocketServerOptions;
+  connectionOptions?: DuplexOptions;
 }
 
 export interface RouteOptions<RawServer extends RawServerBase = RawServerDefault, RawRequest extends RawRequestDefaultExpression<RawServer> = RawRequestDefaultExpression<RawServer>, RawReply extends RawReplyDefaultExpression<RawServer> = RawReplyDefaultExpression<RawServer>, RouteGeneric extends RouteGenericInterface = RouteGenericInterface, ContextConfig = ContextConfigDefault, SchemaCompiler = fastify.FastifySchema> extends fastify.RouteOptions<RawServer, RawRequest, RawReply, RouteGeneric, ContextConfig, SchemaCompiler>, WebsocketRouteOptions<RawServer, RawRequest, RouteGeneric> {}

index.js

@@ -58,7 +58,7 @@ function fastifyWebsocket (fastify, opts, next) {
     wss.handleUpgrade(rawRequest, rawRequest[kWs], rawRequest[kWsHead], (socket) => {
       wss.emit('connection', socket, rawRequest)
 
-      const connection = WebSocket.createWebSocketStream(socket)
+      const connection = WebSocket.createWebSocketStream(socket, opts.connectionOptions)
       connection.socket = socket
 
       connection.socket.on('newListener', event => {

test/base.js

@@ -1,6 +1,7 @@
 'use strict'
 
 const http = require('http')
+const util = require('util')
 const split = require('split2')
 const test = require('tap').test
 const Fastify = require('fastify')
@@ -304,6 +305,46 @@ test('Should be able to pass clientTracking option in false to websocket-stream'
   })
 })
 
+test('Should be able to pass custom connectionOptions to createWebSocketStream', (t) => {
+  t.plan(3)
+
+  const fastify = Fastify()
+  t.teardown(() => fastify.close())
+
+  const connectionOptions = {
+    readableObjectMode: true
+  }
+
+  fastify.register(fastifyWebsocket, { connectionOptions })
+
+  fastify.get('/', { websocket: true }, (connection, request) => {
+    // readableObjectMode was added in Node v12.3.0 so for earlier versions
+    // we check the encapsulated readable state directly
+    const mode = (typeof connection.readableObjectMode === 'undefined')
+      ? connection._readableState.objectMode
+      : connection.readableObjectMode
+    t.equal(mode, true)
+    connection.socket.binaryType = 'arraybuffer'
+
+    connection.once('data', (chunk) => {
+      const message = new util.TextDecoder().decode(chunk)
+      t.equal(message, 'Hello')
+    })
+    t.teardown(() => connection.destroy())
+  })
+
+  fastify.listen(0, (err) => {
+    t.error(err)
+
+    const ws = new WebSocket('ws://localhost:' + fastify.server.address().port)
+    const client = WebSocket.createWebSocketStream(ws, { encoding: 'utf8' })
+    t.teardown(() => client.destroy())
+
+    client.setEncoding('utf8')
+    client.write('Hello')
+  })
+})
+
 test('Should gracefully close with a connected client', (t) => {
   t.plan(6)