diff --git a/src/encoding/armor.js b/src/encoding/armor.js
index 09ecd06e..979e5663 100644
--- a/src/encoding/armor.js
+++ b/src/encoding/armor.js
@@ -283,14 +283,17 @@ function dearmor(input) {
         await stream.pipe(readable, writable, {
           preventClose: true
         });
-        const checksumVerifiedString = await stream.readToEnd(checksumVerified);
         const writer = stream.getWriter(writable);
-        await writer.ready;
-        if (checksum !== checksumVerifiedString && (checksum || config.checksum_required)) {
-          await writer.abort(new Error("Ascii armor integrity check on message failed: '" + checksum + "' should be '" +
-                  checksumVerifiedString + "'"));
-        } else {
+        try {
+          const checksumVerifiedString = await stream.readToEnd(checksumVerified);
+          if (checksum !== checksumVerifiedString && (checksum || config.checksum_required)) {
+            throw new Error("Ascii armor integrity check on message failed: '" + checksum + "' should be '" +
+                    checksumVerifiedString + "'");
+          }
+          await writer.ready;
           await writer.close();
+        } catch(e) {
+          await writer.abort(e);
         }
       });
     } catch(e) {
diff --git a/src/openpgp.js b/src/openpgp.js
index 62fc10f9..1c89b616 100644
--- a/src/openpgp.js
+++ b/src/openpgp.js
@@ -297,9 +297,8 @@ export function encryptKey({ privateKey, passphrase }) {
  * @async
  * @static
  */
-export function encrypt({ data, dataType, publicKeys, privateKeys, passwords, sessionKey, filename, compression=config.compression, armor=true, asStream, detached=false, signature=null, returnSessionKey=false, wildcard=false, date=new Date(), fromUserId={}, toUserId={} }) {
+export function encrypt({ data, dataType, publicKeys, privateKeys, passwords, sessionKey, filename, compression=config.compression, armor=true, asStream=util.isStream(data), detached=false, signature=null, returnSessionKey=false, wildcard=false, date=new Date(), fromUserId={}, toUserId={} }) {
   checkData(data); publicKeys = toArray(publicKeys); privateKeys = toArray(privateKeys); passwords = toArray(passwords);
-  if (asStream === undefined) asStream = util.isStream(data);
 
   if (!nativeAEAD() && asyncProxy) { // use web worker if web crypto apis are not supported
     return asyncProxy.delegate('encrypt', { data, dataType, publicKeys, privateKeys, passwords, sessionKey, filename, compression, armor, asStream, detached, signature, returnSessionKey, wildcard, date, fromUserId, toUserId });
@@ -352,9 +351,8 @@ export function encrypt({ data, dataType, publicKeys, privateKeys, passwords, se
  * @async
  * @static
  */
-export function decrypt({ message, privateKeys, passwords, sessionKeys, publicKeys, format='utf8', asStream, signature=null, date=new Date() }) {
+export function decrypt({ message, privateKeys, passwords, sessionKeys, publicKeys, format='utf8', asStream=message.fromStream, signature=null, date=new Date() }) {
   checkMessage(message); publicKeys = toArray(publicKeys); privateKeys = toArray(privateKeys); passwords = toArray(passwords); sessionKeys = toArray(sessionKeys);
-  if (asStream === undefined) asStream = message.fromStream;
 
   if (!nativeAEAD() && asyncProxy) { // use web worker if web crypto apis are not supported
     return asyncProxy.delegate('decrypt', { message, privateKeys, passwords, sessionKeys, publicKeys, format, asStream, signature, date });
@@ -416,10 +414,9 @@ export function decrypt({ message, privateKeys, passwords, sessionKeys, publicKe
  * @async
  * @static
  */
-export function sign({ data, dataType, privateKeys, armor=true, asStream, detached=false, date=new Date(), fromUserId={} }) {
+export function sign({ data, dataType, privateKeys, armor=true, asStream=util.isStream(data), detached=false, date=new Date(), fromUserId={} }) {
   checkData(data);
   privateKeys = toArray(privateKeys);
-  if (asStream === undefined) asStream = util.isStream(data);
 
   if (asyncProxy) { // use web worker if available
     return asyncProxy.delegate('sign', {
@@ -459,10 +456,9 @@ export function sign({ data, dataType, privateKeys, armor=true, asStream, detach
  * @async
  * @static
  */
-export function verify({ message, publicKeys, asStream, signature=null, date=new Date() }) {
+export function verify({ message, publicKeys, asStream=message.fromStream, signature=null, date=new Date() }) {
   checkCleartextOrMessage(message);
   publicKeys = toArray(publicKeys);
-  if (asStream === undefined) asStream = message.fromStream;
 
   if (asyncProxy) { // use web worker if available
     return asyncProxy.delegate('verify', { message, publicKeys, asStream, signature, date });
@@ -470,10 +466,27 @@ export function verify({ message, publicKeys, asStream, signature=null, date=new
 
   return Promise.resolve().then(async function() {
     const result = {};
-    result.signatures = signature ? await message.verifyDetached(signature, publicKeys, date) : await message.verify(publicKeys, date, asStream);
+    const signatures = signature ? await message.verifyDetached(signature, publicKeys, date) : await message.verify(publicKeys, date, asStream);
     result.data = message instanceof CleartextMessage ? message.getText() : message.getLiteralData();
     result.data = await convertStream(result.data, asStream);
-    result.signatures = stream.readToEnd(result.signatures, arr => arr);
+    if (asStream) {
+      result.data = stream.transformPair(signatures, async (readable, writable) => {
+        const signatures = stream.readToEnd(readable, arr => arr);
+        result.signatures = signatures.catch(() => []);
+        await stream.pipe(result.data, writable, {
+          preventClose: true
+        });
+        const writer = stream.getWriter(writable);
+        try {
+          await signatures;
+          await writer.close();
+        } catch(e) {
+          await writer.abort(e);
+        }
+      });
+    } else {
+      result.signatures = await stream.readToEnd(signatures, arr => arr);
+    }
     return result;
   }).catch(onError.bind(null, 'Error verifying cleartext signed message'));
 }
diff --git a/src/packet/packet.js b/src/packet/packet.js
index 00ab90fd..5eeb11e4 100644
--- a/src/packet/packet.js
+++ b/src/packet/packet.js
@@ -266,7 +266,7 @@ export default {
       return done || !value || !value.length;
     } catch(e) {
       if (writer) {
-        writer.abort(e);
+        await writer.abort(e);
         return true;
       } else {
         throw e;
diff --git a/src/stream.js b/src/stream.js
index 1ecaed4a..719cae01 100644
--- a/src/stream.js
+++ b/src/stream.js
@@ -41,15 +41,19 @@ async function pipe(input, target, options) {
   if (!util.isStream(input)) {
     input = toStream(input);
   }
-  if (input.externalBuffer) {
-    const writer = target.getWriter();
-    for (let i = 0; i < input.externalBuffer.length; i++) {
-      await writer.ready;
-      writer.write(input.externalBuffer[i]);
+  try {
+    if (input.externalBuffer) {
+      const writer = target.getWriter();
+      for (let i = 0; i < input.externalBuffer.length; i++) {
+        await writer.ready;
+        await writer.write(input.externalBuffer[i]);
+      }
+      writer.releaseLock();
     }
-    writer.releaseLock();
+    return await input.pipeTo(target, options).catch(function() {});
+  } catch(e) {
+    util.print_debug_error(e);
   }
-  return input.pipeTo(target, options).catch(function() {});
 }
 
 function transformRaw(input, options) {
@@ -184,6 +188,7 @@ function passiveClone(input) {
               await writer.write(value);
             }
           } catch(e) {
+            controller.error(e);
             await writer.abort(e);
           }
         });
diff --git a/test/general/streaming.js b/test/general/streaming.js
index ae385ffc..5e3e0685 100644
--- a/test/general/streaming.js
+++ b/test/general/streaming.js
@@ -159,6 +159,40 @@ describe('Streaming', function() {
     expect(canceled).to.be.true;
   });
 
+  it('Sign: Input stream should be canceled when canceling encrypted stream', async function() {
+    const privKey = (await openpgp.key.readArmored(priv_key)).keys[0];
+    await privKey.decrypt(passphrase);
+
+    let plaintext = [];
+    let i = 0;
+    let canceled = false;
+    const data = new ReadableStream({
+      async pull(controller) {
+        await new Promise(setTimeout);
+        if (i++ < 10) {
+          let randomBytes = await openpgp.crypto.random.getRandomBytes(1024);
+          controller.enqueue(randomBytes);
+          plaintext.push(randomBytes);
+        } else {
+          controller.close();
+        }
+      },
+      cancel() {
+        canceled = true;
+      }
+    });
+    const encrypted = await openpgp.sign({
+      data,
+      privateKeys: privKey
+    });
+    const reader = openpgp.stream.getReader(encrypted.data);
+    expect(await reader.readBytes(1024)).to.match(/^-----BEGIN PGP MESSAGE-----\r\n/);
+    if (i > 10) throw new Error('Data did not arrive early.');
+    reader.releaseLock();
+    await openpgp.stream.cancel(encrypted.data);
+    expect(canceled).to.be.true;
+  });
+
   it('Encrypt and decrypt larger message roundtrip', async function() {
     let plaintext = [];
     let i = 0;
@@ -418,6 +452,52 @@ describe('Streaming', function() {
     }
   });
 
+  it('Sign/verify: Detect armor checksum error (unsafe_stream=true)', async function() {
+    let unsafe_streamValue = openpgp.config.unsafe_stream;
+    openpgp.config.unsafe_stream = true;
+    try {
+      const pubKey = (await openpgp.key.readArmored(pub_key)).keys[0];
+      const privKey = (await openpgp.key.readArmored(priv_key)).keys[0];
+      await privKey.decrypt(passphrase);
+
+      let plaintext = [];
+      let i = 0;
+      const data = new ReadableStream({
+        async pull(controller) {
+          await new Promise(resolve => setTimeout(resolve, 100));
+          if (i++ < 10) {
+            let randomBytes = await openpgp.crypto.random.getRandomBytes(1024);
+            controller.enqueue(randomBytes);
+            plaintext.push(randomBytes);
+          } else {
+            controller.close();
+          }
+        }
+      });
+      const encrypted = await openpgp.sign({
+        data,
+        privateKeys: privKey
+      });
+
+      const msgAsciiArmored = encrypted.data;
+      const message = await openpgp.message.readArmored(openpgp.stream.transform(msgAsciiArmored, value => {
+        if (value.length > 1000) return value.slice(0, 499) + 'a' + value.slice(500);
+        return value;
+      }));
+      const decrypted = await openpgp.verify({
+        publicKeys: pubKey,
+        message
+      });
+      expect(util.isStream(decrypted.data)).to.be.true;
+      expect(await openpgp.stream.getReader(openpgp.stream.clone(decrypted.data)).readBytes(10)).not.to.deep.equal(plaintext[0]);
+      if (i > 10) throw new Error('Data did not arrive early.');
+      await expect(openpgp.stream.readToEnd(decrypted.data)).to.be.rejectedWith('Ascii armor integrity check on message failed');
+      expect(await decrypted.signatures).to.exist.and.have.length(0);
+    } finally {
+      openpgp.config.unsafe_stream = unsafe_streamValue;
+    }
+  });
+
   it('Encrypt and decrypt larger message roundtrip (draft04)', async function() {
     let aead_protectValue = openpgp.config.aead_protect;
     let aead_chunk_size_byteValue = openpgp.config.aead_chunk_size_byte;
@@ -551,6 +631,59 @@ describe('Streaming', function() {
     }
   });
 
+  it('Sign/verify: Input stream should be canceled when canceling decrypted stream (draft04)', async function() {
+    let aead_protectValue = openpgp.config.aead_protect;
+    let aead_chunk_size_byteValue = openpgp.config.aead_chunk_size_byte;
+    openpgp.config.aead_protect = true;
+    openpgp.config.aead_chunk_size_byte = 4;
+    try {
+      const pubKey = (await openpgp.key.readArmored(pub_key)).keys[0];
+      const privKey = (await openpgp.key.readArmored(priv_key)).keys[0];
+      await privKey.decrypt(passphrase);
+
+      let plaintext = [];
+      let i = 0;
+      let canceled = false;
+      const data = new ReadableStream({
+        async pull(controller) {
+          await new Promise(resolve => setTimeout(resolve, 10));
+          if (i++ < 10) {
+            let randomBytes = await openpgp.crypto.random.getRandomBytes(1024);
+            controller.enqueue(randomBytes);
+            plaintext.push(randomBytes);
+          } else {
+            controller.close();
+          }
+        },
+        cancel() {
+          canceled = true;
+        }
+      });
+      const encrypted = await openpgp.sign({
+        data,
+        privateKeys: privKey
+      });
+
+      const msgAsciiArmored = encrypted.data;
+      const message = await openpgp.message.readArmored(msgAsciiArmored);
+      const decrypted = await openpgp.verify({
+        publicKeys: pubKey,
+        message
+      });
+      expect(util.isStream(decrypted.data)).to.be.true;
+      const reader = openpgp.stream.getReader(decrypted.data);
+      expect(await reader.readBytes(1024)).to.deep.equal(plaintext[0]);
+      if (i > 10) throw new Error('Data did not arrive early.');
+      reader.releaseLock();
+      await openpgp.stream.cancel(decrypted.data, new Error('canceled by test'));
+      expect(canceled).to.be.true;
+      expect(await decrypted.signatures).to.exist.and.have.length(0);
+    } finally {
+      openpgp.config.aead_protect = aead_protectValue;
+      openpgp.config.aead_chunk_size_byte = aead_chunk_size_byteValue;
+    }
+  });
+
   it("Don't pull entire input stream when we're not pulling encrypted stream", async function() {
     let plaintext = [];
     let i = 0;
@@ -577,6 +710,36 @@ describe('Streaming', function() {
     expect(i).to.be.lessThan(50);
   });
 
+  it("Sign: Don't pull entire input stream when we're not pulling signed stream", async function() {
+    const pubKey = (await openpgp.key.readArmored(pub_key)).keys[0];
+    const privKey = (await openpgp.key.readArmored(priv_key)).keys[0];
+    await privKey.decrypt(passphrase);
+
+    let plaintext = [];
+    let i = 0;
+    const data = new ReadableStream({
+      async pull(controller) {
+        if (i++ < 100) {
+          let randomBytes = await openpgp.crypto.random.getRandomBytes(1024);
+          controller.enqueue(randomBytes);
+          plaintext.push(randomBytes);
+        } else {
+          controller.close();
+        }
+        await new Promise(setTimeout);
+      }
+    });
+    const encrypted = await openpgp.sign({
+      data,
+      privateKeys: privKey
+    });
+    const reader = openpgp.stream.getReader(encrypted.data);
+    expect(await reader.readBytes(1024)).to.match(/^-----BEGIN PGP MESSAGE-----\r\n/);
+    if (i > 10) throw new Error('Data did not arrive early.');
+    await new Promise(resolve => setTimeout(resolve, 3000));
+    expect(i).to.be.lessThan(50);
+  });
+
   it("Don't pull entire input stream when we're not pulling decrypted stream (draft04)", async function() {
     let aead_protectValue = openpgp.config.aead_protect;
     let aead_chunk_size_byteValue = openpgp.config.aead_chunk_size_byte;
@@ -619,4 +782,50 @@ describe('Streaming', function() {
       openpgp.config.aead_chunk_size_byte = aead_chunk_size_byteValue;
     }
   });
+
+  it("Sign/verify: Don't pull entire input stream when we're not pulling verified stream (draft04)", async function() {
+    let aead_protectValue = openpgp.config.aead_protect;
+    let aead_chunk_size_byteValue = openpgp.config.aead_chunk_size_byte;
+    openpgp.config.aead_protect = true;
+    openpgp.config.aead_chunk_size_byte = 4;
+    try {
+      const pubKey = (await openpgp.key.readArmored(pub_key)).keys[0];
+      const privKey = (await openpgp.key.readArmored(priv_key)).keys[0];
+      await privKey.decrypt(passphrase);
+
+      let plaintext = [];
+      let i = 0;
+      const data = new ReadableStream({
+        async pull(controller) {
+          if (i++ < 100) {
+            let randomBytes = await openpgp.crypto.random.getRandomBytes(1024);
+            controller.enqueue(randomBytes);
+            plaintext.push(randomBytes);
+          } else {
+            controller.close();
+          }
+          await new Promise(setTimeout);
+        }
+      });
+      const encrypted = await openpgp.sign({
+        data,
+        privateKeys: privKey
+      });
+      const msgAsciiArmored = encrypted.data;
+      const message = await openpgp.message.readArmored(msgAsciiArmored);
+      const decrypted = await openpgp.verify({
+        publicKeys: pubKey,
+        message
+      });
+      expect(util.isStream(decrypted.data)).to.be.true;
+      const reader = openpgp.stream.getReader(decrypted.data);
+      expect(await reader.readBytes(1024)).to.deep.equal(plaintext[0]);
+      if (i > 10) throw new Error('Data did not arrive early.');
+      await new Promise(resolve => setTimeout(resolve, 3000));
+      expect(i).to.be.lessThan(50);
+    } finally {
+      openpgp.config.aead_protect = aead_protectValue;
+      openpgp.config.aead_chunk_size_byte = aead_chunk_size_byteValue;
+    }
+  });
 });