Merge pull request #416 from jgilbert01/s3-head-object

S3 head object command

Author: barnettet31

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "aws-lambda-stream",
-  "version": "1.1.6",
+  "version": "1.1.7",
   "description": "Create stream processors with AWS Lambda functions.",
   "keywords": [
     "aws",

package.json

@@ -3,7 +3,7 @@
 import { Readable } from 'stream';
 import {
   CopyObjectCommand,
-  DeleteObjectCommand, GetObjectCommand, ListObjectsV2Command, PutObjectCommand, S3Client,
+  DeleteObjectCommand, GetObjectCommand, HeadObjectCommand, ListObjectsV2Command, PutObjectCommand, S3Client,
 } from '@aws-sdk/client-s3';
 import { NodeHttpHandler } from '@smithy/node-http-handler';
 import Promise from 'bluebird';
@@ -47,6 +47,15 @@ class Connector {
     return this.clients[pipelineId];
   }
 
+  headObject(inputParams, ctx) {
+    const params = {
+      Bucket: this.bucketName,
+      ...inputParams,
+    };
+
+    return this._sendCommand(new HeadObjectCommand(params), ctx);
+  }
+
   putObject(inputParams, ctx) {
     const params = {
       Bucket: this.bucketName,

src/connectors/s3.js

@@ -193,3 +193,32 @@ export const pageObjectsFromS3 = ({
     .map(listObjects)
     .parallel(parallel);
 };
+
+export const headS3Object = ({
+  id: pipelineId,
+  debug = d('s3'),
+  bucketName = process.env.BUCKET_NAME,
+  headRequestField = 'headRequest',
+  headResponseField = 'headResponse',
+  parallel = Number(process.env.S3_PARALLEL) || Number(process.env.PARALLEL) || 8,
+  step = 'get',
+  ...opt
+} = {}) => {
+  const connector = new Connector({
+    pipelineId, debug, bucketName, ...opt,
+  });
+
+  const headObject = (uow) => {
+    if (!uow[headRequestField]) return _(Promise.resolve(uow));
+
+    const p = () => connector.headObject(uow[headRequestField], uow)
+      .then((headResponse) => ({ ...uow, [headResponseField]: headResponse }))
+      .catch(rejectWithFault(uow));
+
+    return _(uow.metrics?.w(p, step) || p()); // wrap promise in a stream
+  };
+
+  return (s) => s
+    .map(headObject)
+    .parallel(parallel);
+};

src/queries/s3.js

@@ -6,7 +6,7 @@ import { Readable } from 'stream';
 import { mockClient } from 'aws-sdk-client-mock';
 import {
   CopyObjectCommand,
-  DeleteObjectCommand, GetObjectCommand, ListObjectsV2Command, PutObjectCommand, S3Client,
+  DeleteObjectCommand, GetObjectCommand, HeadObjectCommand, ListObjectsV2Command, PutObjectCommand, S3Client,
 } from '@aws-sdk/client-s3';
 import { sdkStreamMixin } from '@smithy/util-stream';
 
@@ -199,4 +199,20 @@ describe('connectors/s3.js', () => {
     });
     expect(data).to.deep.equal({});
   });
+  it('should head object', async () => {
+    const spy = sinon.spy(() => ({}));
+    mockS3.on(HeadObjectCommand).callsFake(spy);
+    const inputParams = {
+      Key: 'k1',
+    };
+    const data = await new Connector({
+      debug: debug('s3'),
+      bucketName: 'b1',
+    }).headObject(inputParams);
+    expect(spy).to.have.been.calledWith({
+      Key: 'k1',
+      Bucket: 'b1',
+    });
+    expect(data).to.deep.equal({});
+  });
 });

test/unit/connectors/s3.test.js

@@ -12,6 +12,7 @@ import {
   toGetObjectRequest,
   toGetObjectRequest2,
   getObjectFromS3AsStream,
+  headS3Object,
 } from '../../../src/queries/s3';
 
 import Connector from '../../../src/connectors/s3';
@@ -329,6 +330,68 @@ describe('queries/s3.js', () => {
       })
       .done(done);
   });
+  it('should head object', (done) => {
+    const stub = sinon.stub(Connector.prototype, 'headObject').resolves({
+      Metadata: {
+        testkey: '1',
+      },
+    });
+
+    const uows = [{
+      headRequest: {
+        Key: 'k1',
+      },
+    }];
+
+    _(uows)
+      .through(headS3Object())
+      .collect()
+      .tap((collected) => {
+        // console.log(JSON.stringify(collected, null, 2));
+
+        expect(stub).to.have.been.calledWith({
+          Key: 'k1',
+        });
+
+        expect(collected.length).to.equal(1);
+        expect(collected[0]).to.deep.equal({
+          headRequest: {
+            Key: 'k1',
+          },
+          headResponse: {
+            Metadata: {
+              testkey: '1',
+            },
+          },
+        });
+      })
+      .done(done);
+  });
+  it('should head object missing headRequestField', (done) => {
+    sinon.stub(Connector.prototype, 'headObject').resolves({
+      Metadata: {
+        testkey: '1',
+      },
+    });
+
+    const uows = [{
+      // headRequest: {
+      //   Key: 'k1',
+      // },
+    }];
+
+    _(uows)
+      .through(headS3Object())
+      .collect()
+      .tap((collected) => {
+        // console.log(JSON.stringify(collected, null, 2));
+
+        expect(collected[0]).to.deep.equal({});
+
+        expect(collected.length).to.equal(1);
+      })
+      .done(done);
+  });
 });
 
 const GET_RESPONSE = {