diff --git a/README.md b/README.md
index d199404..ec0b690 100644
--- a/README.md
+++ b/README.md
@@ -14,6 +14,7 @@ Validate an incoming data stream using SHACL. If the incoming data is valid, it
 - `incoming`: channel which is used as the data source.
 - `outgoing`: channel into which valid data is written.
 - `report`: an optional channel into which the SHACL reports of invalid input data is written. (default: `null`)
+- `validationIsFatal`: throw a fatal error if validation fails. (default: `false`)
 
 ## Limitations
 
diff --git a/src/error.ts b/src/error.ts
index f66958f..de5c23d 100644
--- a/src/error.ts
+++ b/src/error.ts
@@ -1,6 +1,7 @@
 enum ShaclErrorType {
     FileSystemError,
     InvalidRdfFormat,
+    ValidationFailed,
 }
 
 export class ShaclError extends Error {
@@ -25,4 +26,11 @@ export class ShaclError extends Error {
             ShaclErrorType.InvalidRdfFormat,
         );
     }
+
+    static validationFailed() {
+        return new ShaclError(
+            "Validation failed and is fatal.",
+            ShaclErrorType.ValidationFailed,
+        );
+    }
 }
diff --git a/src/index.ts b/src/index.ts
index 22fbfac..27419cd 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -11,13 +11,13 @@ type ValidateArguments = {
     incoming: Stream<string>;
     outgoing: Writer<string>;
     report?: Writer<string>;
-    verbose?: boolean;
+    validationIsFatal?: boolean;
 };
 
 export async function validate(
     args: ValidateArguments,
 ): Promise<() => Promise<void>> {
-    const { path, incoming, outgoing, report } = args;
+    const { path, incoming, outgoing, report, validationIsFatal } = args;
 
     // Initialize the shared serializer.
     const prefixes = new PrefixMapFactory().prefixMap();
@@ -63,6 +63,8 @@ export async function validate(
             // Pass through data if valid.
             if (result.conforms) {
                 await outgoing.push(data);
+            } else if (validationIsFatal) {
+                throw ShaclError.validationFailed();
             } else if (report) {
                 const resultRaw = serializer.transform(result.dataset);
                 await report.push(resultRaw);
diff --git a/tests/error.test.ts b/tests/error.test.ts
index 56ad0fc..0698da3 100644
--- a/tests/error.test.ts
+++ b/tests/error.test.ts
@@ -2,6 +2,11 @@ import { describe, test, expect } from "vitest";
 import { validate } from "../src";
 import { SimpleStream } from "@ajuvercr/js-runner";
 import { ShaclError } from "../src/error";
+import fs from "fs";
+
+const shaclPath = "./tests/shacl/point.ttl";
+
+const invalidRdfData = fs.readFileSync("./tests/data/invalid.ttl").toString();
 
 describe("errors", () => {
     test("invalid shacl file path", async () => {
@@ -27,4 +32,45 @@ describe("errors", () => {
 
         expect(func).rejects.toThrow(ShaclError.invalidRdfFormat());
     });
+
+    test("invalid input data", async () => {
+        expect.assertions(1);
+
+        const incoming = new SimpleStream<string>();
+        const outgoing = new SimpleStream<string>();
+
+        const func = await validate({
+            path: shaclPath,
+            incoming,
+            outgoing,
+        });
+        await func();
+
+        expect(
+            incoming.push("This is not a valid Turtle file!"),
+        ).rejects.toThrow(ShaclError.invalidRdfFormat());
+
+        await incoming.end();
+        await outgoing.end();
+    });
+
+    test("invalid and fatal", async () => {
+        expect.assertions(1);
+
+        const incoming = new SimpleStream<string>();
+
+        const func = await validate({
+            path: shaclPath,
+            incoming,
+            outgoing: new SimpleStream<string>(),
+            validationIsFatal: true,
+        });
+        await func();
+
+        expect(incoming.push(invalidRdfData)).rejects.toThrow(
+            ShaclError.validationFailed(),
+        );
+
+        await incoming.end();
+    });
 });