Fix bad content parsing results (#902)

* Send headers after computing

* Send headers after computing

Author: giurgiur99

src/components/c2d/compute_engine_base.ts

@@ -73,7 +73,7 @@ export abstract class C2DEngine {
     consumerAddress: string,
     jobId: string,
     index: number
-  ): Promise<Readable>
+  ): Promise<{ stream: Readable; headers: any }>
 
   public abstract cleanupExpiredStorage(job: DBComputeJob): Promise<boolean>
 

src/components/c2d/compute_engine_docker.ts

@@ -421,7 +421,7 @@ export class C2DEngineDocker extends C2DEngine {
     consumerAddress: string,
     jobId: string,
     index: number
-  ): Promise<Readable> {
+  ): Promise<{ stream: Readable; headers: any }> {
     const jobs = await this.db.getJob(jobId, null, consumerAddress)
     if (jobs.length === 0) {
       return null
@@ -430,14 +430,24 @@ export class C2DEngineDocker extends C2DEngine {
     for (const i of results) {
       if (i.index === index) {
         if (i.type === 'algorithmLog') {
-          return createReadStream(
-            this.getC2DConfig().tempFolder + '/' + jobId + '/data/logs/algorithm.log'
-          )
+          return {
+            stream: createReadStream(
+              this.getC2DConfig().tempFolder + '/' + jobId + '/data/logs/algorithm.log'
+            ),
+            headers: {
+              'Content-Type': 'text/plain'
+            }
+          }
         }
         if (i.type === 'output') {
-          return createReadStream(
-            this.getC2DConfig().tempFolder + '/' + jobId + '/data/outputs/outputs.tar'
-          )
+          return {
+            stream: createReadStream(
+              this.getC2DConfig().tempFolder + '/' + jobId + '/data/outputs/outputs.tar'
+            ),
+            headers: {
+              'Content-Type': 'application/octet-stream'
+            }
+          }
         }
       }
     }

src/components/c2d/compute_engine_opf_k8.ts

@@ -276,7 +276,7 @@ export class C2DEngineOPFK8 extends C2DEngine {
     consumerAddress: string,
     jobId: string,
     index: number
-  ): Promise<Readable> {
+  ): Promise<{ stream: Readable; headers: any }> {
     const nonce: number = new Date().getTime()
     const config = await getConfiguration()
     // signature check on operator service is only owner + jobId
@@ -307,7 +307,10 @@ export class C2DEngineOPFK8 extends C2DEngine {
         const message = `Exception on getComputeJobResult. Status: ${response.status}, ${response.statusText}`
         throw new Error(message)
       }
-      return response.data
+      return {
+        stream: response.data,
+        headers: response.headers
+      }
     } catch (e) {
       console.error(e)
     }

src/components/core/compute/getResults.ts

@@ -90,16 +90,15 @@ export class ComputeGetResultHandler extends CommandHandler {
         jobId,
         task.index
       )
-      const anyResp: any = respStream as any
       const response: P2PCommandResponse = {
-        stream: respStream,
+        stream: respStream.stream,
         status: {
           httpStatus: 200
         }
       }
       // need to pass the headers properly
-      if (anyResp.headers) {
-        response.status.headers = anyResp.headers
+      if (respStream.headers) {
+        response.status.headers = respStream.headers
       }
       return response
     } catch (error) {