feat: upload and download stream file

__tests__/lib/pg/service.test.js

@@ -1,5 +1,6 @@
 const cds = require('@sap/cds')
 const deploy = require('@sap/cds/lib/dbs/cds-deploy')
+const { fs } = require('@sap/cds/lib/utils/cds-utils')
 // const path = require('path')
 
 // mock (package|.cds'rc).json entries
@@ -236,6 +237,27 @@ describe.each(suiteEnvironments)(
  ibu: 20
  })
  })
+
+ test('odata: stream -> sql: upload/dowload stream Media Content', async () => {
+ //Create a media
+ const response = await request.post('/beershop/TypeChecks').send({
+ type_mediaType: 'text/plain'
+ })
+
+ const entry = JSON.parse(response.text)
+
+ const mediaPath = `/beershop/TypeChecks(${entry.ID})/type_mediaContent`
+
+ const buffer = fs.readFileSync('__tests__/__assets__/test.sql')
+
+ // Upload file
+ await request.put(mediaPath).attach('file', buffer).expect(204)
+
+ // Download file
+ const mediaResponse = await request.get(mediaPath).send().responseType('blob').expect(200)
+
+ expect(mediaResponse.body.toString().includes(buffer.toString())).toEqual(true)
+ })
  })
 
  describe('odata: GET on Draft enabled Entity -> sql: SELECT', () => {

lib/pg/execute.js

@@ -5,6 +5,8 @@ const { postProcess, getPostProcessMapper } = require('@sap/cds/libx/_runtime/db
 const { PG_TYPE_CONVERSION_MAP } = require('./converters/conversion')
 const { flattenArray } = require('./utils/deep')
 const { remapColumnNames } = require('./utils/columns')
+const { Readable } = require('stream')
+
 const DEBUG = cds.debug('cds-pg|sql')
 
 /*
@@ -23,14 +25,17 @@ const DEBUG = cds.debug('cds-pg|sql')
  * @param {*} txTimestamp
  * @return {import('pg').QueryArrayResult}
  */
-const executeGenericCQN = (model, dbc, query, user, locale, txTimestamp) => {
+const executeGenericCQN = async (model, dbc, query, user, locale, txTimestamp) => {
  const { sql, values = [] } = _cqnToSQL(model, query, user, locale, txTimestamp)
+
+ const vals = await _convertStreamValues(values)
+
  if (/^\s*insert/i.test(sql)) {
  return executeInsertCQN(model, dbc, query, user, locale, txTimestamp)
  }
  const isOne = query.SELECT && query.SELECT.one
  const postPropertyMapper = getPostProcessMapper(PG_TYPE_CONVERSION_MAP, model, query)
- return _executeSQLReturningRows(dbc, sql, values, isOne, postPropertyMapper)
+ return _executeSQLReturningRows(dbc, sql, vals, isOne, postPropertyMapper)
 }
 
 /**
@@ -231,12 +236,61 @@ const executeUpdateCQN = async (model, dbc, cqn, user, locale, txTimestamp) => {
  return Array.isArray(result) ? result.length : result
 }
 
+/**
+ * Processes a SELECT CQN statement and executes the query against the database.
+ * The result rows are processed and returned.
+ * @param {Object} model
+ * @param {import('pg').PoolClient} dbc
+ * @param {Object} query
+ * @param {*} user
+ * @param {String} locale
+ * @param {*} txTimestamp
+ * @return {import('pg-query-stream').QueryStream}
+ */
+const executeSelectStreamCQN = async (model, dbc, query, user, locale, txTimestamp) => {
+ const result = await executeSelectCQN(model, dbc, query, user, locale, txTimestamp)
+
+ if (result == null || result.length === 0) {
+ return
+ }
+
+ let val = Array.isArray(result) ? Object.values(result[0])[0] : Object.values(result)[0]
+ if (val === null) {
+ return null
+ }
+ if (typeof val === 'number') {
+ val = val.toString()
+ }
+
+ const stream_ = Readable.from(val)
+
+ return { value: stream_ }
+}
+
+const _convertStreamValues = async (values) => {
+ let any
+ values.forEach((v, i) => {
+ if (v && typeof v.pipe === 'function') {
+ any = values[i] = new Promise((resolve) => {
+ const chunks = []
+ v.on('data', (chunk) => chunks.push(chunk))
+ v.on('end', () => resolve(Buffer.concat(chunks)))
+ v.on('error', () => {
+ v.removeAllListeners('error')
+ v.push(null)
+ })
+ })
+ }
+ })
+ return any ? Promise.all(values) : values
+}
+
 module.exports = {
  delete: executeGenericCQN,
  insert: executeInsertCQN,
  update: executeUpdateCQN,
  read: executeSelectCQN,
- //stream: executeSelectStreamCQN,
+ stream: executeSelectStreamCQN,
  cqn: executeGenericCQN,
  sql: executePlainSQL
 }