added reactive repository and reactive entity manager

package.json

@@ -49,6 +49,7 @@
     "path": "^0.12.7",
     "reflect-metadata": "^0.1.3",
     "require-all": "^2.0.0",
+    "rxjs": "^5.0.0-beta.7",
     "sha1": "^1.1.1"
   },
   "scripts": {

sample/sample1-simple-entity/app.ts

@@ -19,6 +19,7 @@ createConnection(options).then(connection => {
     let post = new Post();
     post.text = "Hello how are you?";
     post.title = "hello";
+    post.likesCount = 100;
 
     let postRepository = connection.getRepository<Post>(Post);
 

sample/sample15-we-are-reactive/app.ts

@@ -0,0 +1,31 @@
+import {createConnection, CreateConnectionOptions} from "../../src/typeorm";
+import {Post} from "./entity/Post";
+
+const options: CreateConnectionOptions = {
+    driver: "mysql",
+    connection: {
+        host: "192.168.99.100",
+        port: 3306,
+        username: "root",
+        password: "admin",
+        database: "test",
+        autoSchemaCreate: true
+    },
+    entities: [Post]
+};
+
+createConnection(options).then(connection => {
+
+    let post = new Post();
+    post.text = "Hello how are you?";
+    post.title = "hello";
+    post.likesCount = 0;
+
+    let postRepository = connection.getReactiveRepository(Post);
+
+    postRepository
+        .persist(post)
+        .do(post => console.log(post.title + " stream!"))
+        .subscribe(post => console.log("Post has been saved"));
+
+}, error => console.log("Cannot connect: ", error));

sample/sample15-we-are-reactive/entity/Post.ts

@@ -0,0 +1,19 @@
+import {PrimaryColumn, Column} from "../../../src/columns";
+import {Table} from "../../../src/tables";
+
+@Table("sample15_post")
+export class Post {
+
+    @PrimaryColumn("int", { generated: true })
+    id: number;
+
+    @Column()
+    title: string;
+
+    @Column()
+    text: string;
+
+    @Column()
+    likesCount: number;
+
+}

src/connection/Connection.ts

@@ -18,11 +18,13 @@ import {NoConnectionForRepositoryError} from "./error/NoConnectionForRepositoryE
 import {CannotImportAlreadyConnectedError} from "./error/CannotImportAlreadyConnectedError";
 import {CannotCloseNotConnectedError} from "./error/CannotCloseNotConnectedError";
 import {CannotConnectAlreadyConnectedError} from "./error/CannotConnectAlreadyConnectedError";
+import {ReactiveRepository} from "../repository/ReactiveRepository";
+import {ReactiveEntityManager} from "../repository/ReactiveEntityManager";
 
 /**
  * Temporary type to store and link both repository and its metadata.
  */
-type RepositoryAndMetadata = { repository: Repository<any>, metadata: EntityMetadata };
+type RepositoryAndMetadata = { metadata: EntityMetadata, repository: Repository<any>, reactiveRepository: ReactiveRepository<any> };
 
 /**
  * A single connection instance to the database. Each connection has its own repositories, subscribers and metadatas.
@@ -44,6 +46,11 @@ export class Connection {
      */
     readonly entityManager: EntityManager;
 
+    /**
+     * Gets ReactiveEntityManager of this connection.
+     */
+    readonly reactiveEntityManager: ReactiveEntityManager;
+
     /**
      * The name of the connection.
      */
@@ -113,6 +120,7 @@ export class Connection {
         this.driver.connectionOptions = options;
         this.options = options;
         this.entityManager = new EntityManager(this);
+        this.reactiveEntityManager = new ReactiveEntityManager(this);
     }
 
     // -------------------------------------------------------------------------
@@ -245,6 +253,21 @@ export class Connection {
         return repoMeta.repository;
     }
 
+    /**
+     * Gets repository for the given entity class.
+     */
+    getReactiveRepository<Entity>(entityClass: ConstructorFunction<Entity>|Function): ReactiveRepository<Entity> {
+        if (!this.isConnected)
+            throw new NoConnectionForRepositoryError(this.name);
+
+        const metadata = this.entityMetadatas.findByTarget(entityClass);
+        const repoMeta = this.repositoryAndMetadatas.find(repoMeta => repoMeta.metadata === metadata);
+        if (!repoMeta)
+            throw new RepositoryNotFoundError(this.name, entityClass);
+
+        return repoMeta.reactiveRepository;
+    }
+
     // -------------------------------------------------------------------------
     // Private Methods
     // -------------------------------------------------------------------------
@@ -298,9 +321,11 @@ export class Connection {
      * Creates a temporary object RepositoryAndMetadata to store relation between repository and metadata.
      */
     private createRepoMeta(metadata: EntityMetadata): RepositoryAndMetadata {
+        const repository = new Repository<any>(this, this.entityMetadatas, metadata);
         return {
             metadata: metadata,
-            repository: new Repository<any>(this, this.entityMetadatas, metadata)
+            repository: repository,
+            reactiveRepository: new ReactiveRepository(repository)
         };
     }
 

src/repository/EntityManager.ts

@@ -3,6 +3,7 @@ import {QueryBuilder} from "../query-builder/QueryBuilder";
 import {FindOptions} from "./FindOptions";
 import {Repository} from "./Repository";
 import {ConstructorFunction} from "../common/ConstructorFunction";
+import {ReactiveRepository} from "./ReactiveRepository";
 
 /**
  * Entity manager supposed to work with any entity, automatically find its repository and call its method, whatever
@@ -27,6 +28,13 @@ export class EntityManager {
     getRepository<Entity>(entityClass: ConstructorFunction<Entity>|Function): Repository<Entity> {
         return this.connection.getRepository(entityClass);
     }
+
+    /**
+     * Gets reactive repository of the given entity.
+     */
+    getReactiveRepository<Entity>(entityClass: ConstructorFunction<Entity>|Function): ReactiveRepository<Entity> {
+        return this.connection.getReactiveRepository(entityClass);
+    }
     
     /**
      * Checks if entity has an id.

src/repository/ReactiveEntityManager.ts

@@ -0,0 +1,235 @@
+import {Connection} from "../connection/Connection";
+import {QueryBuilder} from "../query-builder/QueryBuilder";
+import {FindOptions} from "./FindOptions";
+import {Repository} from "./Repository";
+import {ConstructorFunction} from "../common/ConstructorFunction";
+import {ReactiveRepository} from "./ReactiveRepository";
+import * as Rx from "rxjs/Rx";
+
+/**
+ * Entity manager supposed to work with any entity, automatically find its repository and call its method, whatever
+ * entity type are you passing. This version of ReactiveEntityManager works with reactive streams and observables.
+ */
+export class ReactiveEntityManager {
+
+    // -------------------------------------------------------------------------
+    // Constructor
+    // -------------------------------------------------------------------------
+
+    constructor(private connection: Connection) {
+    }
+
+    // -------------------------------------------------------------------------
+    // Public Methods
+    // -------------------------------------------------------------------------
+
+    /**
+     * Gets repository of the given entity.
+     */
+    getRepository<Entity>(entityClass: ConstructorFunction<Entity>|Function): Repository<Entity> {
+        return this.connection.getRepository(entityClass);
+    }
+
+    /**
+     * Gets reactive repository of the given entity.
+     */
+    getReactiveRepository<Entity>(entityClass: ConstructorFunction<Entity>|Function): ReactiveRepository<Entity> {
+        return this.connection.getReactiveRepository(entityClass);
+    }
+    
+    /**
+     * Checks if entity has an id.
+     */
+    hasId(entity: Function): boolean {
+        return this.getReactiveRepository(entity.constructor).hasId(entity);
+    }
+
+    /**
+     * Creates a new query builder that can be used to build an sql query.
+     */
+    createQueryBuilder<Entity>(entityClass: ConstructorFunction<Entity>|Function, alias: string): QueryBuilder<Entity> {
+        return this.getReactiveRepository(entityClass).createQueryBuilder(alias);
+    }
+
+    /**
+     * Creates a new entity. If fromRawEntity is given then it creates a new entity and copies all entity properties
+     * from this object into a new entity (copies only properties that should be in a new entity).
+     */
+    create<Entity>(entityClass: ConstructorFunction<Entity>|Function, fromRawEntity?: Object): Entity {
+        return this.getReactiveRepository(entityClass).create(fromRawEntity);
+    }
+
+    /**
+     * Creates a entities from the given array of plain javascript objects.
+     */
+    createMany<Entity>(entityClass: ConstructorFunction<Entity>|Function, copyFromObjects: any[]): Entity[] {
+        return this.getReactiveRepository(entityClass).createMany(copyFromObjects);
+    }
+
+    /**
+     * Creates a new entity from the given plan javascript object. If entity already exist in the database, then
+     * it loads it (and everything related to it), replaces all values with the new ones from the given object
+     * and returns this new entity. This new entity is actually a loaded from the db entity with all properties
+     * replaced from the new object.
+     */
+    initialize<Entity>(entityClass: ConstructorFunction<Entity>|Function, object: Object): Rx.Observable<Entity> {
+        return this.getReactiveRepository(entityClass).initialize(object);
+    }
+
+    /**
+     * Merges two entities into one new entity.
+     */
+    merge<Entity>(entity1: Entity, entity2: Entity): Entity {
+        return <Entity> this.getReactiveRepository(<any> entity1).merge(entity1, entity2);
+    }
+
+    /**
+     * Persists (saves) a given entity in the database.
+     */
+    persist<Entity>(entity: Entity): Rx.Observable<Entity> {
+        // todo: extra casting is used strange tsc error here, check later maybe typescript bug
+        return <any> this.getReactiveRepository(<any> entity.constructor).persist(entity);
+    }
+
+    /**
+     * Removes a given entity from the database.
+     */
+    remove<Entity>(entity: Entity) {
+        return this.getReactiveRepository(<any> entity.constructor).remove(entity);
+    }
+
+    /**
+     * Finds entities that match given conditions.
+     */
+    find<Entity>(entityClass: ConstructorFunction<Entity>|Function): Rx.Observable<Entity[]>;
+
+    /**
+     * Finds entities that match given conditions.
+     */
+    find<Entity>(entityClass: ConstructorFunction<Entity>|Function, conditions: Object): Rx.Observable<Entity[]>;
+
+    /**
+     * Finds entities that match given conditions.
+     */
+    find<Entity>(entityClass: ConstructorFunction<Entity>|Function, options: FindOptions): Rx.Observable<Entity[]>;
+
+    /**
+     * Finds entities that match given conditions.
+     */
+    find<Entity>(entityClass: ConstructorFunction<Entity>|Function, conditions: Object, options: FindOptions): Rx.Observable<Entity[]>;
+
+    /**
+     * Finds entities that match given conditions.
+     */
+    find<Entity>(entityClass: ConstructorFunction<Entity>|Function, conditionsOrFindOptions?: Object|FindOptions, options?: FindOptions): Rx.Observable<Entity[]> {
+        if (conditionsOrFindOptions && options) {
+            return this.getReactiveRepository(entityClass).find(conditionsOrFindOptions, options);
+            
+        } else if (conditionsOrFindOptions) {
+            return this.getReactiveRepository(entityClass).find(conditionsOrFindOptions);
+            
+        } else {
+            return this.getReactiveRepository(entityClass).find();
+        }
+    }
+
+    /**
+     * Finds entities that match given conditions.
+     */
+    findAndCount<Entity>(entityClass: ConstructorFunction<Entity>|Function): Rx.Observable<[ Entity[], number ]>;
+
+    /**
+     * Finds entities that match given conditions.
+     */
+    findAndCount<Entity>(entityClass: ConstructorFunction<Entity>|Function, conditions: Object): Rx.Observable<[ Entity[], number ]>;
+
+    /**
+     * Finds entities that match given conditions.
+     */
+    findAndCount<Entity>(entityClass: ConstructorFunction<Entity>|Function, options: FindOptions): Rx.Observable<[ Entity[], number ]>;
+
+    /**
+     * Finds entities that match given conditions.
+     */
+    findAndCount<Entity>(entityClass: ConstructorFunction<Entity>|Function, conditions: Object, options: FindOptions): Rx.Observable<[ Entity[], number ]>;
+
+    /**
+     * Finds entities that match given conditions.
+     */
+    findAndCount<Entity>(entityClass: ConstructorFunction<Entity>|Function, conditionsOrFindOptions?: Object|FindOptions, options?: FindOptions): Rx.Observable<[Entity[], number]> {
+        if (conditionsOrFindOptions && options) {
+            return this.getReactiveRepository(entityClass).findAndCount(conditionsOrFindOptions, options);
+
+        } else if (conditionsOrFindOptions) {
+            return this.getReactiveRepository(entityClass).findAndCount(conditionsOrFindOptions);
+
+        } else {
+            return this.getReactiveRepository(entityClass).findAndCount();
+        }
+    }
+
+    /**
+     * Finds first entity that matches given conditions.
+     */
+    findOne<Entity>(entityClass: ConstructorFunction<Entity>|Function): Rx.Observable<Entity>;
+
+    /**
+     * Finds first entity that matches given conditions.
+     */
+    findOne<Entity>(entityClass: ConstructorFunction<Entity>|Function, conditions: Object): Rx.Observable<Entity>;
+
+    /**
+     * Finds first entity that matches given conditions.
+     */
+    findOne<Entity>(entityClass: ConstructorFunction<Entity>|Function, options: FindOptions): Rx.Observable<Entity>;
+
+    /**
+     * Finds first entity that matches given conditions.
+     */
+    findOne<Entity>(entityClass: ConstructorFunction<Entity>|Function, conditions: Object, options: FindOptions): Rx.Observable<Entity>;
+
+    /**
+     * Finds first entity that matches given conditions.
+     */
+    findOne<Entity>(entityClass: ConstructorFunction<Entity>|Function, conditionsOrFindOptions?: Object|FindOptions, options?: FindOptions): Rx.Observable<Entity> {
+        if (conditionsOrFindOptions && options) {
+            return this.getReactiveRepository(entityClass).findOne(conditionsOrFindOptions, options);
+
+        } else if (conditionsOrFindOptions) {
+            return this.getReactiveRepository(entityClass).findOne(conditionsOrFindOptions);
+
+        } else {
+            return this.getReactiveRepository(entityClass).findOne();
+        }
+    }
+
+    /**
+     * Finds entity with given id.
+     */
+    findOneById<Entity>(entityClass: ConstructorFunction<Entity>|Function, id: any, options?: FindOptions): Rx.Observable<Entity> {
+        return this.getReactiveRepository(entityClass).findOneById(id, options);
+    }
+
+    /**
+     * Executes raw SQL query and returns raw database results.
+     */
+    query(query: string): Rx.Observable<any> {
+        return Rx.Observable.fromPromise(this.connection.driver.query(query));
+    }
+
+    /**
+     * Wraps given function execution (and all operations made there) in a transaction.
+     */
+    transaction(runInTransaction: () => Promise<any>): Rx.Observable<any> {
+        let runInTransactionResult: any;
+        return Rx.Observable.fromPromise(this.connection.driver
+            .beginTransaction()
+            .then(() => runInTransaction())
+            .then(result => {
+                runInTransactionResult = result;
+                return this.connection.driver.endTransaction();
+            })
+            .then(() => runInTransactionResult));
+    }
+
+}

src/repository/ReactiveRepository.ts

@@ -0,0 +1,211 @@
+import {QueryBuilder} from "../query-builder/QueryBuilder";
+import {FindOptions} from "./FindOptions";
+import {Repository} from "./Repository";
+import * as Rx from "rxjs/Rx";
+
+/**
+ * Repository is supposed to work with your entity objects. Find entities, insert, update, delete, etc.
+ * This version of Repository is using rxjs library and Observables instead of promises.
+ */
+export class ReactiveRepository<Entity> {
+    
+    // -------------------------------------------------------------------------
+    // Constructor
+    // -------------------------------------------------------------------------
+
+    constructor(private repository: Repository<Entity>) {
+
+    }
+
+    // -------------------------------------------------------------------------
+    // Public Methods
+    // -------------------------------------------------------------------------
+
+    /**
+     * Checks if entity has an id.
+     */
+    hasId(entity: Entity): boolean {
+        return this.repository.hasId(entity);
+    }
+
+    /**
+     * Creates a new query builder that can be used to build a sql query.
+     */
+    createQueryBuilder(alias: string): QueryBuilder<Entity> {
+        return this.repository.createQueryBuilder(alias);
+    }
+
+    /**
+     * Creates a new entity. If fromRawEntity is given then it creates a new entity and copies all entity properties
+     * from this object into a new entity (copies only properties that should be in a new entity).
+     */
+    create(fromRawEntity?: Object): Entity {
+        return this.repository.create(fromRawEntity);
+    }
+
+    /**
+     * Creates entities from a given array of plain javascript objects.
+     */
+    createMany(copyFromObjects: Object[]): Entity[] {
+        return this.repository.createMany(copyFromObjects);
+    }
+
+    /**
+     * Creates a new entity from the given plan javascript object. If entity already exist in the database, then
+     * it loads it (and everything related to it), replaces all values with the new ones from the given object
+     * and returns this new entity. This new entity is actually a loaded from the db entity with all properties
+     * replaced from the new object.
+     */
+    initialize(object: Object): Rx.Observable<Entity> {
+        return Rx.Observable.fromPromise(this.repository.initialize(object));
+    }
+
+    /**
+     * Merges two entities into one new entity.
+     */
+    merge(entity1: Entity, entity2: Entity): Entity {
+        return this.repository.merge(entity1, entity2);
+    }
+
+    /**
+     * Persists (saves) a given entity in the database. If entity does not exist in the database then it inserts it, 
+     * else if entity already exist in the database then it updates it.
+     */
+    persist(entity: Entity): Rx.Observable<Entity> {
+        return Rx.Observable.fromPromise(this.repository.persist(entity));
+    }
+
+    /**
+     * Removes a given entity from the database.
+     */
+    remove(entity: Entity): Rx.Observable<Entity> {
+        return Rx.Observable.fromPromise(this.repository.remove(entity));
+    }
+
+    /**
+     * Finds all entities.
+     */
+    find(): Rx.Observable<Entity[]>;
+
+    /**
+     * Finds entities that match given conditions.
+     */
+    find(conditions: Object): Rx.Observable<Entity[]>;
+
+    /**
+     * Finds entities with .
+     */
+    find(options: FindOptions): Rx.Observable<Entity[]>;
+
+    /**
+     * Finds entities that match given conditions.
+     */
+    find(conditions: Object, options: FindOptions): Rx.Observable<Entity[]>;
+
+    /**
+     * Finds entities that match given conditions.
+     */
+    find(conditionsOrFindOptions?: Object|FindOptions, options?: FindOptions): Rx.Observable<Entity[]> {
+        if (conditionsOrFindOptions && options) {
+            return Rx.Observable.fromPromise(this.repository.find(conditionsOrFindOptions, options));
+
+        } else if (conditionsOrFindOptions) {
+            return Rx.Observable.fromPromise(this.repository.find(conditionsOrFindOptions));
+
+        } else {
+            return Rx.Observable.fromPromise(this.repository.find());
+        }
+    }
+
+    /**
+     * Finds entities that match given conditions.
+     */
+    findAndCount(): Rx.Observable<[ Entity[], number ]>;
+
+    /**
+     * Finds entities that match given conditions.
+     */
+    findAndCount(conditions: Object): Rx.Observable<[ Entity[], number ]>;
+
+    /**
+     * Finds entities that match given conditions.
+     */
+    findAndCount(options: FindOptions): Rx.Observable<[ Entity[], number ]>;
+
+    /**
+     * Finds entities that match given conditions.
+     */
+    findAndCount(conditions: Object, options: FindOptions): Rx.Observable<[ Entity[], number ]>;
+
+    /**
+     * Finds entities that match given conditions.
+     */
+    findAndCount(conditionsOrFindOptions?: Object|FindOptions, options?: FindOptions): Rx.Observable<[ Entity[], number ]> {
+        if (conditionsOrFindOptions && options) {
+            return Rx.Observable.fromPromise(this.repository.findAndCount(conditionsOrFindOptions, options));
+
+        } else if (conditionsOrFindOptions) {
+            return Rx.Observable.fromPromise(this.repository.findAndCount(conditionsOrFindOptions));
+
+        } else {
+            return Rx.Observable.fromPromise(this.repository.findAndCount());
+        }
+    }
+
+    /**
+     * Finds first entity that matches given conditions.
+     */
+    findOne(): Rx.Observable<Entity>;
+
+    /**
+     * Finds first entity that matches given conditions.
+     */
+    findOne(conditions: Object): Rx.Observable<Entity>;
+
+    /**
+     * Finds first entity that matches given conditions.
+     */
+    findOne(options: FindOptions): Rx.Observable<Entity>;
+
+    /**
+     * Finds first entity that matches given conditions.
+     */
+    findOne(conditions: Object, options: FindOptions): Rx.Observable<Entity>;
+
+    /**
+     * Finds first entity that matches given conditions.
+     */
+    findOne(conditionsOrFindOptions?: Object|FindOptions, options?: FindOptions): Rx.Observable<Entity> {
+        if (conditionsOrFindOptions && options) {
+            return Rx.Observable.fromPromise(this.repository.findOne(conditionsOrFindOptions, options));
+
+        } else if (conditionsOrFindOptions) {
+            return Rx.Observable.fromPromise(this.repository.findOne(conditionsOrFindOptions));
+
+        } else {
+            return Rx.Observable.fromPromise(this.repository.findOne());
+        }
+    }
+
+    /**
+     * Finds entity with given id.
+     */
+    findOneById(id: any, options?: FindOptions): Rx.Observable<Entity> {
+        return Rx.Observable.fromPromise(this.repository.findOneById(id, options));
+    }
+
+    /**
+     * Executes raw SQL query and returns raw database results.
+     */
+    query(query: string): Rx.Observable<any> {
+        return Rx.Observable.fromPromise(this.repository.query(query));
+    }
+
+    /**
+     * Wraps given function execution (and all operations made there) in a transaction.
+     */
+    transaction(runInTransaction: () => Promise<any>): Rx.Observable<any> {
+        return Rx.Observable.fromPromise(this.repository.transaction(runInTransaction));
+    }
+
+}