'Common implementation for tests where the underlying producer throws
an exception when its resumeProducing is called.'

def resumeProducingRaises(self, consumer, expectedExceptions):

    class ThrowingProducer(NonStreamingProducer, ):
        def resumeProducing(self):
            if (self.counter == 2):
                return (1 / 0)
            else:
                NonStreamingProducer.resumeProducing(self)
    nsProducer = ThrowingProducer(consumer)
    streamingProducer = _PullToPush(nsProducer, consumer)
    consumer.registerProducer(streamingProducer, True)
    loggedMsgs = []
    log.addObserver(loggedMsgs.append)
    self.addCleanup(log.removeObserver, loggedMsgs.append)
    def unregister(orig=consumer.unregisterProducer):
        orig()
        streamingProducer.stopStreaming()
    consumer.unregisterProducer = unregister
    streamingProducer.startStreaming()
    done = streamingProducer._coopTask.whenDone()
    done.addErrback((lambda reason: reason.trap(TaskStopped)))
    def stopped(ign):
        self.assertEqual(consumer.value(), '01')
        errors = self.flushLoggedErrors()
        self.assertEqual(len(errors), len(expectedExceptions))
        for (f, (expected, msg), logMsg) in zip(errors, expectedExceptions, loggedMsgs):
            self.assertTrue(f.check(expected))
            self.assertIn(msg, logMsg['why'])
        self.assertTrue(streamingProducer._finished)
    done.addCallback(stopped)
    return done
