diff --git a/types/n3/index.d.ts b/types/n3/index.d.ts
index 3ceb2c6003..eba19244fb 100644
--- a/types/n3/index.d.ts
+++ b/types/n3/index.d.ts
@@ -151,7 +151,7 @@ export interface StreamParserConstructor {
 }
 export const StreamParser: StreamParserConstructor;
 
-export interface N3StreamParser<Q extends BaseQuad = Quad> extends RDF.Stream<Q>, NodeJS.WritableStream, RDF.Sink<Q> {
+export interface N3StreamParser<Q extends BaseQuad = Quad> extends RDF.Stream<Q>, NodeJS.WritableStream, RDF.Sink<EventEmitter, RDF.Stream<Q>> {
     // Below are the NodeJS.ReadableStream methods,
     // we can not extend the interface directly,
     // as `read` clashes with RDF.Sink.
@@ -199,7 +199,7 @@ export interface StreamWriterConstructor {
 }
 export const StreamWriter: StreamWriterConstructor;
 
-export interface N3StreamWriter<Q extends RDF.BaseQuad = Quad> extends NodeJS.ReadWriteStream, RDF.Sink<Q> {}
+export interface N3StreamWriter<Q extends RDF.BaseQuad = Quad> extends NodeJS.ReadWriteStream, RDF.Sink<RDF.Stream<Q>, EventEmitter> {}
 
 export interface N3Store<Q_RDF extends RDF.BaseQuad = RDF.Quad, Q_N3 extends BaseQuad = Quad> extends RDF.Store<Q_RDF> {
     readonly size: number;
diff --git a/types/rdf-ext/index.d.ts b/types/rdf-ext/index.d.ts
index d1179b1933..fb4660666e 100644
--- a/types/rdf-ext/index.d.ts
+++ b/types/rdf-ext/index.d.ts
@@ -7,20 +7,21 @@
 import { Sink } from 'rdf-js';
 import DataFactory = require('./lib/DataFactory');
 import EventEmitter = require('events');
+import { Stream } from 'stream';
 
-type SinkMap = {
-  find(mediaType: string): Sink;
+type SinkMap<InputStream extends EventEmitter, OutputStream extends EventEmitter> = {
+  find(mediaType: string): Sink<InputStream, OutputStream>;
   import(mediaType: string, input: any, options: any): any;
   list(): string[];
 } & {
-  [mediaType: string]: Sink;
+  [mediaType: string]: Sink<InputStream, OutputStream>;
 };
 
 declare class DataFactoryExt extends DataFactory {
   static asEvent: (p: any) => EventEmitter;
   static waitFor: (event: any) => Promise<any>;
-  static Parsers: SinkMap;
-  static Serializers: SinkMap;
+  static Parsers: SinkMap<EventEmitter, Stream>;
+  static Serializers: SinkMap<Stream, EventEmitter>;
 }
 
 export = DataFactoryExt;
diff --git a/types/rdf-ext/rdf-ext-tests.ts b/types/rdf-ext/rdf-ext-tests.ts
index fa0b0a43b8..11e6c59604 100644
--- a/types/rdf-ext/rdf-ext-tests.ts
+++ b/types/rdf-ext/rdf-ext-tests.ts
@@ -1,7 +1,10 @@
 import rdf = require('rdf-ext');
-import { Literal, Quad, Dataset, NamedNode } from 'rdf-js';
+import { Literal, Quad, Dataset, NamedNode, Stream, Sink } from 'rdf-js';
 import QuadExt = require('rdf-ext/lib/Quad');
 import DataFactoryExt = require('rdf-ext/lib/DataFactory');
+import DatasetExt = require('rdf-ext/lib/Dataset');
+import { EventEmitter } from 'events';
+import { Readable } from 'stream';
 
 function NamedNode_toCanonical(): string {
     const iri = 'http://example.org';
@@ -279,3 +282,11 @@ function Dataset_toJSON() {
         && quad.graph !== null;
     });
 }
+
+async function dataset_parserImport() {
+    const dataset: DatasetExt = <any> {};
+    const parserSink: Sink<EventEmitter, Stream> = <any> {};
+    const stream: Readable = <any> {};
+
+    const promise: DatasetExt = await dataset.import(parserSink.import(stream));
+}
diff --git a/types/rdf-js/index.d.ts b/types/rdf-js/index.d.ts
index 119b60aa3a..a9f40705d6 100644
--- a/types/rdf-js/index.d.ts
+++ b/types/rdf-js/index.d.ts
@@ -365,7 +365,7 @@ export interface Source<Q extends BaseQuad = Quad> {
  *
  * For example parsers, serializers, transformations and stores can implement the Sink interface.
  */
-export interface Sink<Q extends BaseQuad = Quad> {
+export interface Sink<InputStream extends EventEmitter, OutputStream extends EventEmitter> {
     /**
      * Consumes the given stream.
      *
@@ -376,7 +376,7 @@ export interface Sink<Q extends BaseQuad = Quad> {
      * @param stream The stream that will be consumed.
      * @return The resulting event emitter.
      */
-    import(stream: Stream<Q>): EventEmitter;
+    import(stream: InputStream): OutputStream;
 }
 
 /**
@@ -387,7 +387,7 @@ export interface Sink<Q extends BaseQuad = Quad> {
  *
  * Access to stores LDP or SPARQL endpoints can be implemented with a Store inteface.
  */
-export interface Store<Q extends BaseQuad = Quad> extends Source<Q>, Sink<Q> {
+export interface Store<Q extends BaseQuad = Quad> extends Source<Q>, Sink<Stream<Q>, EventEmitter> {
     /**
      * Removes all streamed quads.
      *
diff --git a/types/rdf-js/rdf-js-tests.ts b/types/rdf-js/rdf-js-tests.ts
index bf29c33c0a..c76c9c79d7 100644
--- a/types/rdf-js/rdf-js-tests.ts
+++ b/types/rdf-js/rdf-js-tests.ts
@@ -109,13 +109,13 @@ function test_stream() {
     const matchStream8: Stream = source.match(term, term, term, term);
     const matchStream9: Stream = source.match(term, term, term, /.*/);
 
-    const sink: Sink = <any> {};
+    const sink: Sink<Stream, EventEmitter> = <any> {};
     const graph: Quad_Graph = <any> {};
     const eventEmitter1: EventEmitter = sink.import(stream);
 
     const store: Store = <any> {};
     const storeSource: Source = store;
-    const storeSink: Sink = store;
+    const storeSink: Sink<Stream, EventEmitter> = store;
     const eventEmitter2: EventEmitter = store.remove(stream);
     const eventEmitter3: EventEmitter = store.removeMatches();
     const eventEmitter4: EventEmitter = store.removeMatches(term);
diff --git a/types/rdfjs__parser-jsonld/index.d.ts b/types/rdfjs__parser-jsonld/index.d.ts
index a562e37d00..62fabb04f2 100644
--- a/types/rdfjs__parser-jsonld/index.d.ts
+++ b/types/rdfjs__parser-jsonld/index.d.ts
@@ -5,6 +5,7 @@
 
 import { Context } from 'jsonld/jsonld-spec';
 import { DataFactory, Sink, Stream, BaseQuad, Quad } from 'rdf-js';
+import { EventEmitter } from 'events';
 
 declare namespace Parser {
     interface ParserOptions {
@@ -14,10 +15,10 @@ declare namespace Parser {
     }
 }
 
-declare class Parser<Q extends BaseQuad = Quad> implements Sink<Q> {
+declare class Parser<Q extends BaseQuad = Quad> implements Sink<EventEmitter, Stream<Q>> {
     constructor(options?: Parser.ParserOptions);
 
-    import(stream: Stream<Q>, options?: Parser.ParserOptions): Stream<Q>;
+    import(stream: EventEmitter, options?: Parser.ParserOptions): Stream<Q>;
 }
 
 export = Parser;
diff --git a/types/rdfjs__parser-jsonld/rdfjs__parser-jsonld-tests.ts b/types/rdfjs__parser-jsonld/rdfjs__parser-jsonld-tests.ts
index 250718b2d2..dae259cd6c 100644
--- a/types/rdfjs__parser-jsonld/rdfjs__parser-jsonld-tests.ts
+++ b/types/rdfjs__parser-jsonld/rdfjs__parser-jsonld-tests.ts
@@ -1,6 +1,7 @@
 import Parser = require('@rdfjs/parser-jsonld');
 import { Context } from 'jsonld/jsonld-spec';
 import { DataFactory, Sink, Stream, BaseQuad } from 'rdf-js';
+import { EventEmitter } from 'events';
 
 const baseIRI = '';
 const context: Context = {} as any;
@@ -11,7 +12,7 @@ const parser1 = new Parser();
 const parser2 = new Parser({});
 const parser3 = new Parser({ baseIRI, context, factory });
 
-const sink: Sink = parser1;
+const sink: Sink<EventEmitter, Stream> = parser1;
 
 const eventEmitter1: Stream = parser1.import(stream);
 const eventEmitter2: Stream = parser1.import(stream, {});
diff --git a/types/rdfjs__parser-n3/index.d.ts b/types/rdfjs__parser-n3/index.d.ts
index 7efd1916a6..d5f7d35d5d 100644
--- a/types/rdfjs__parser-n3/index.d.ts
+++ b/types/rdfjs__parser-n3/index.d.ts
@@ -4,16 +4,17 @@
 // Definitions: https://github.com/DefinitelyTyped/DefinitelyTyped
 
 import { Sink, Stream, DataFactory, BaseQuad, Quad } from 'rdf-js';
+import { EventEmitter } from 'events';
 
 interface ParserOptions {
     baseIRI?: string;
     factory?: DataFactory;
 }
 
-declare class Parser<Q extends BaseQuad = Quad> implements Sink<Q> {
+declare class Parser<Q extends BaseQuad = Quad> implements Sink<EventEmitter, Stream<Q>> {
     constructor(options?: ParserOptions);
 
-    import(stream: Stream<Q>, options?: ParserOptions): Stream<Q>;
+    import(stream: EventEmitter, options?: ParserOptions): Stream<Q>;
 }
 
 export = Parser;
diff --git a/types/rdfjs__parser-n3/rdfjs__parser-n3-tests.ts b/types/rdfjs__parser-n3/rdfjs__parser-n3-tests.ts
index 94bbab1002..a9b2215d42 100644
--- a/types/rdfjs__parser-n3/rdfjs__parser-n3-tests.ts
+++ b/types/rdfjs__parser-n3/rdfjs__parser-n3-tests.ts
@@ -1,5 +1,6 @@
 import ParserN3 = require('@rdfjs/parser-n3');
 import { Stream, DataFactory, Sink, BaseQuad } from 'rdf-js';
+import { EventEmitter } from 'events';
 
 const factory: DataFactory = <any> {};
 const baseIRI = '';
@@ -9,7 +10,7 @@ const parser1 = new ParserN3({});
 const parser2 = new ParserN3({ factory });
 const parser3 = new ParserN3({ baseIRI });
 
-const sink: Sink = parser;
+const sink: Sink<EventEmitter, Stream> = parser;
 
 const input: Stream = <any> {};
 const output: Stream = parser.import(input);
diff --git a/types/rdfjs__serializer-jsonld-ext/index.d.ts b/types/rdfjs__serializer-jsonld-ext/index.d.ts
index bf46386750..3a768d57c0 100644
--- a/types/rdfjs__serializer-jsonld-ext/index.d.ts
+++ b/types/rdfjs__serializer-jsonld-ext/index.d.ts
@@ -19,7 +19,7 @@ declare namespace Serializer {
     }
 }
 
-declare class Serializer<Q extends BaseQuad = Quad> implements Sink<Q> {
+declare class Serializer<Q extends BaseQuad = Quad> implements Sink<Stream<Q>, EventEmitter> {
     constructor(options?: Serializer.SerializerOptions);
 
     import(stream: Stream<Q>, options?: Serializer.SerializerOptions): EventEmitter;
diff --git a/types/rdfjs__serializer-jsonld-ext/rdfjs__serializer-jsonld-ext-tests.ts b/types/rdfjs__serializer-jsonld-ext/rdfjs__serializer-jsonld-ext-tests.ts
index 30f4f5e74f..8925b2eb2f 100644
--- a/types/rdfjs__serializer-jsonld-ext/rdfjs__serializer-jsonld-ext-tests.ts
+++ b/types/rdfjs__serializer-jsonld-ext/rdfjs__serializer-jsonld-ext-tests.ts
@@ -17,9 +17,9 @@ const serializer3 = new Serializer({
     skipContext: true,
     skipGraphProperty: true
 });
-const serializer4 = new Serializer({ encoding: 'object' });
+const serializer4: Serializer = new Serializer({ encoding: 'object' });
 
-const sink: Sink = serializer1;
+const sink: Sink<Stream, EventEmitter> = serializer1;
 
 const eventEmitter1: EventEmitter = serializer1.import(stream);
 const eventEmitter2: EventEmitter = serializer1.import(stream, {});
diff --git a/types/rdfjs__serializer-jsonld/index.d.ts b/types/rdfjs__serializer-jsonld/index.d.ts
index 9b8667f2ef..f8823f2d2e 100644
--- a/types/rdfjs__serializer-jsonld/index.d.ts
+++ b/types/rdfjs__serializer-jsonld/index.d.ts
@@ -12,7 +12,7 @@ declare namespace Serializer {
     }
 }
 
-declare class Serializer<Q extends BaseQuad = Quad> implements Sink<Q> {
+declare class Serializer<Q extends BaseQuad = Quad> implements Sink<Stream<Q>, EventEmitter> {
     constructor(options?: Serializer.SerializerOptions);
 
     import(stream: Stream<Q>, options?: Serializer.SerializerOptions): EventEmitter;
diff --git a/types/rdfjs__serializer-jsonld/rdfjs__serializer-jsonld-tests.ts b/types/rdfjs__serializer-jsonld/rdfjs__serializer-jsonld-tests.ts
index 5651621dd2..b918fc1d2a 100644
--- a/types/rdfjs__serializer-jsonld/rdfjs__serializer-jsonld-tests.ts
+++ b/types/rdfjs__serializer-jsonld/rdfjs__serializer-jsonld-tests.ts
@@ -9,7 +9,7 @@ const serializer2 = new Serializer({});
 const serializer3 = new Serializer({ encoding: 'string' });
 const serializer4 = new Serializer({ encoding: 'object' });
 
-const sink: Sink = serializer1;
+const sink: Sink<Stream, EventEmitter> = serializer1;
 
 const eventEmitter1: EventEmitter = serializer1.import(stream);
 const eventEmitter2: EventEmitter = serializer1.import(stream, {});