Created
March 26, 2015 14:24
-
-
Save codedmart/efd198a59d109758beb0 to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| overriding value connection in trait Rql of type com.scalaRethinkdb.Connection; | |
| value connection has incompatible type | |
| val connection = connect _ |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| trait Rql { | |
| val connection: Connection | |
| def get(db:String, table: String): String = { | |
| // Table example | |
| val query = Json.array( | |
| jNumber(1), Json.array( | |
| jNumber(15), Json.array( | |
| jString(table) | |
| ) | |
| ), | |
| Json.obj( | |
| "db" -> Json.array( | |
| jNumber(14), Json.array( | |
| jString(db) | |
| ) | |
| ) | |
| ) | |
| ) | |
| connection.run(query) | |
| } | |
| } | |
| object R extends Rql { | |
| val DEFAULT_HOST = "localhost" | |
| val DEFAULT_PORT_DRIVER = 28015 | |
| val connection = connect _ | |
| def connect(host: String = DEFAULT_HOST, port: Int = DEFAULT_PORT_DRIVER, db: String = null, auth: String = null): Connection = { | |
| val address = new InetSocketAddress(host, port) | |
| val dbName = getOption[String](db) | |
| val authKey = getOption[String](auth) | |
| Connection(address, dbName, authKey) | |
| } | |
| } | |
| case class Connection(address: InetSocketAddress, dbName: Option[String], authKey: Option[String]) { | |
| private var socket: Socket = _ | |
| private var in: InputStream = _ | |
| private var out: OutputStream = _ | |
| private val token: AtomicInteger = new AtomicInteger() | |
| def nextToken = token.incrementAndGet | |
| reconnect() | |
| def reconnect(): this.type = { | |
| if (isOpen) close() | |
| Try(new Socket(address.getAddress, address.getPort)) match { | |
| case Success(sock) => { | |
| socket = new Socket(address.getAddress, address.getPort) | |
| socket.setKeepAlive(true) | |
| socket.setTcpNoDelay(true) | |
| in = socket.getInputStream() | |
| out = socket.getOutputStream() | |
| handshake() | |
| } | |
| case _ => { | |
| val message = "Could not connect to %s.".format(address) | |
| throw new RethinkDbConnectionError(message) | |
| } | |
| } | |
| def handshake() { | |
| val version = Wire.to(VersionDummy.V0_3) | |
| val protocol = Wire.to(VersionDummy.JSON) | |
| val auth = authKey match { | |
| case None => "" | |
| case Some(str) => str | |
| } | |
| val authLength = authKey match { | |
| case None => 0 | |
| case Some(str) => str.length | |
| } | |
| val message = pack(version) ++ pack(authLength) ++ auth.getBytes ++ pack(protocol) | |
| out.write(message) | |
| out.flush() | |
| val response = recvNullTerminatedString() | |
| if ("SUCCESS" != new String(response, "US-ASCII")) { | |
| throw new RethinkDbConnectionError("Error establishing driver handshake") | |
| } | |
| } | |
| this | |
| } | |
| def close(): Unit = { | |
| if (!isOpen) println("Connection is closed.") | |
| in.close() | |
| out.close() | |
| socket.close() | |
| } | |
| // TODO: This is not how this should actually be, just using for testing purposes | |
| def run(query: argonaut.Json): String = { | |
| val token = nextToken | |
| val queryLength = query.toString.length | |
| val message = pack(token, 8) ++ pack(queryLength) ++ query.toString.getBytes | |
| out.write(message) | |
| out.flush() | |
| val resToken = recvAll(8) | |
| val resLength = recvAll(4) | |
| val resMessage = recvAll(unpack(resLength)) | |
| new String(resMessage, "US-ASCII") | |
| } | |
| def isOpen: Boolean = { | |
| socket != null && !socket.isClosed && socket.isConnected | |
| } | |
| private def recvAll(length: Int): Array[Byte] = { | |
| Stream.continually(in.read()) | |
| .take(length) | |
| .map(_.toByte) | |
| .toArray | |
| } | |
| private def recvNullTerminatedString(): Array[Byte] = { | |
| Stream.continually(in.read()) | |
| .takeWhile(_ != 0) | |
| .map(_.toByte) | |
| .toArray | |
| } | |
| } |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment