simple refactoring, upgrade CLI

2016-07-26 10:55:28 +03:00
parent c2fea4048c
commit 8fd1f45010
7 changed files with 166 additions and 98 deletions
@@ -5,16 +5,14 @@ import io.netty.channel.nio.NioEventLoopGroup
 import io.netty.channel.socket.nio.NioServerSocketChannel
 import io.netty.handler.codec.http.*
 import objects.Car
-import java.io.ByteArrayInputStream
 import java.io.ByteArrayOutputStream
-import java.util.*
 import kotlin.concurrent.thread

 /**
 * Created by user on 7/6/16.
 */

-//хардкод это плохо, но пока так...:)
+val timeDeltaToDrop = 600000//time in ms. if car is inactive more than this time, server drop session with car
 val carServerPort: Int = 7925
 val webServerPort: Int = 7926
 val handlerThreadsCount: Int = 100
@@ -25,8 +23,103 @@ val connectUrl = "/connect"

 fun main(args: Array<String>) {

-    println("car server started")
-    val serverCarThread = thread {
+    val carServer = getCarServerThread()
+//    val webServer = getWebServerThread()
+    val carsDestroy = getCarsDestroyThread()
+    carServer.start()
+    carsDestroy.start()
+
+    //CL user interface
+    val helpString = "available commands:\n" +
+            "cars - get list of connected cars\n" +
+            "route [car_id] - setting a route for car with car id.\n" +
+            "refloc - refresh all car locations\n" +
+            "stop - exit from this interface and stop all servers\n"
+    println(helpString)
+    val routeRegex = Regex("route [0-9]{1,10}")
+    val environment = objects.Environment.instance
+    while (true) {
+        val readedString = readLine()
+        if (readedString == null) {
+            break
+        }
+        if (readedString.equals("cars", true)) {
+            synchronized(environment, {
+                println(environment.map.values)
+            })
+        } else if (routeRegex.matches(readedString)) {
+
+            val id: Int
+            try {
+                id = readedString.split(" ")[1].toInt()
+            } catch (e: NumberFormatException) {
+                e.printStackTrace()
+                println("error in converting id to int type")
+                println(helpString)
+                continue
+            }
+            val car: Car? =
+                    synchronized(environment, {
+                        environment.map.get(id)
+                    })
+            if (car != null) {
+                println("print way points in polar coordinate als [distance] [rotation angle] in metres and degrees. after enter all points print \"done\". for reset route print \"reset\"")
+                println("e.g. for move from (x,y) to (x+1,y) and back to (x,y) u need enter: 1 0[press enter] 1 180[press enter] done")
+                val routeBuilder = RouteRequest.BuilderRouteRequest()
+                while (true) {
+                    val wayPointInputString = readLine()!!
+                    if (wayPointInputString.equals("reset", true)) {
+                        break
+                    } else if (wayPointInputString.equals("done", true)) {
+                        val requestBytes = ByteArrayOutputStream()
+                        routeBuilder.build().writeTo(CodedOutputStream(requestBytes))
+                        val request = DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, setRouteUrl, Unpooled.copiedBuffer(requestBytes.toByteArray()))
+                        request.headers().set(HttpHeaderNames.HOST, car.host)
+                        request.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE)
+                        request.headers().setInt(HttpHeaderNames.CONTENT_LENGTH, request.content().readableBytes())
+                        Client.sendRequest(request, car.host, car.port, id)
+                        break
+                    } else {
+                        val wayPointData = wayPointInputString.split(" ")
+                        val distance: Double
+                        val angle: Double
+                        try {
+                            distance = wayPointData[0].toDouble()
+                            angle = wayPointData[1].toDouble()
+                        } catch (e: NumberFormatException) {
+                            println("error in convertion angle or distance to double. try again")
+                            continue
+                        }
+                        val wayPoint = RouteRequest.WayPoint.BuilderWayPoint().setDistance(distance).setAngle_delta(angle).build()
+                        routeBuilder.addWayPoint(wayPoint)
+                    }
+                }
+            } else {
+                println("car with id=$id not found")
+            }
+        } else if (readedString.equals("refloc", true)) {
+            val cars = synchronized(environment, { environment.map.values })
+            for (car in cars) {
+                val request = DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, getLocationUrl, Unpooled.EMPTY_BUFFER)
+                request.headers().set(HttpHeaderNames.HOST, car.host)
+                request.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE)
+                request.headers().setInt(HttpHeaderNames.CONTENT_LENGTH, request.content().readableBytes())
+                Client.sendRequest(request, car.host, car.port, car.uid)
+            }
+        } else if (readedString.equals("quit")) {
+            break
+        } else {
+            println("not supported command: $readedString")
+            println(helpString)
+        }
+    }
+    carsDestroy.interrupt()
+    carServer.interrupt()
+}
+
+fun getCarServerThread(): Thread {
+    return thread(false, false, null, "carServer", -1, {
+        println("car server started")
        val bossGroup = NioEventLoopGroup(1)
        val workerGroup = NioEventLoopGroup()
        val b = ServerBootstrap()
@@ -38,82 +131,60 @@ fun main(args: Array<String>) {
            val channel = b.bind(carServerPort).sync().channel()
            channel.closeFuture().sync()
        } catch (e: InterruptedException) {
-            println("car server stoped!")
-            e.printStackTrace()
+            println("car server stoped")
        } finally {
            bossGroup.shutdownGracefully()
            workerGroup.shutdownGracefully()
        }
-    }
+    })
+}

-    //    val serverWebThread = thread {
-    //        val bossGroup = NioEventLoopGroup(1)
-    //        val workerGroup = NioEventLoopGroup()
-    //        val b = ServerBootstrap()
-    //        b.group(bossGroup, workerGroup)
-    //                .channel(NioServerSocketChannel().javaClass)
-    //                .childHandler(web.server.Initializer(handlerThreadsCount))
-    //
-    //        try {
-    //            val channel = b.bind(webServerPort).sync().channel()
-    //            channel.closeFuture().sync()
-    //        } catch (e: InterruptedException) {
-    //            println("web server stoped!")
-    //        } finally {
-    //            bossGroup.shutdownGracefully()
-    //            workerGroup.shutdownGracefully()
-    //        }
-    //    }
-    val serverWebThread = thread {
-        val environment = objects.Environment.instance
-        val scanner = Scanner(System.`in`)
-        while (scanner.hasNext()) {
-            val s = scanner.nextLine()
-            if (s.equals("cars", true)) {
-                synchronized(environment, {
-                    println(environment.map.values)
-                })
-            } else if (s.equals("pathto", true)) {
-                println("print datas. format: [string id] [double distance] [double angle]")
-                val data = scanner.nextLine()
-                val datas = data.split(" ")
-                try {
-                    val id = datas[0].toInt()
-                    val distance = datas[1].toDouble()
-                    val angle = datas[2].toDouble()
+fun getWebServerThread(): Thread {
+    return thread(false, false, null, "webServer", -1, {
+        println("web server started")
+        val bossGroup = NioEventLoopGroup(1)
+        val workerGroup = NioEventLoopGroup()
+        val b = ServerBootstrap()
+        b.group(bossGroup, workerGroup)
+                .channel(NioServerSocketChannel().javaClass)
+                .childHandler(web.server.Initializer(handlerThreadsCount))

-                    val car: Car? =
-                            synchronized(environment, {
-                                environment.map.get(id)
-                            })
-                    if (car != null) {
-                        val wayPoint = RouteRequest.WayPoint.BuilderWayPoint().setDistance(distance).setAngle_delta(angle).build()
-                        val route = RouteRequest.BuilderRouteRequest().addWayPoint(wayPoint).build()
-                        val requestBytes = ByteArrayOutputStream()
-                        route.writeTo(CodedOutputStream(requestBytes))
-                        val request = DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, setRouteUrl, Unpooled.copiedBuffer(requestBytes.toByteArray()))
-                        request.headers().set(HttpHeaderNames.HOST, car.host)
-                        request.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE)
-                        request.headers().setInt(HttpHeaderNames.CONTENT_LENGTH, request.content().readableBytes())
-                        Client.sendRequest(request, car.host, car.port, id)
+        try {
+            val channel = b.bind(webServerPort).sync().channel()
+            channel.closeFuture().sync()
+        } catch (e: InterruptedException) {
+            println("web server stoped")
+        } finally {
+            bossGroup.shutdownGracefully()
+            workerGroup.shutdownGracefully()
+        }
+    })
+}
+
+//thread, that dropped inactive car. every minute check all connected cars and if car is inactive more, als timeDeltaToDrop then remove this car.
+fun getCarsDestroyThread(): Thread {
+    return thread(false, false, null, "dropCar", -1, {
+        var stoped = false
+        while (!stoped) {
+            val environment = objects.Environment.instance
+            synchronized(environment, {
+                val currentTime = System.currentTimeMillis()
+                val keysToRemove = mutableListOf<Int>();
+                for (keyValue in environment.map) {
+                    if ((keyValue.value.lastAction + timeDeltaToDrop) < currentTime) {
+                        keysToRemove.add(keyValue.key)
                    }
-
-                } catch (e: Exception) {
-                    e.printStackTrace()
-                    println("incorrect format")
                }
-            } else if (s.equals("refloc", true)) {
-                val cars = synchronized(environment, { environment.map.values })
-                for (car in cars) {
-                    val request = DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, getLocationUrl, Unpooled.EMPTY_BUFFER)
-                    request.headers().set(HttpHeaderNames.HOST, car.host)
-                    request.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE)
-                    request.headers().setInt(HttpHeaderNames.CONTENT_LENGTH, request.content().readableBytes())
-                    Client.sendRequest(request, car.host, car.port, car.uid)
+                for (key in keysToRemove) {
+                    environment.map.remove(key)
                }
+            })
+            try {
+                Thread.sleep(60000)
+            } catch (e: InterruptedException) {
+                println("thread for destroy cars stoped")
+                stoped = true
            }
        }
-    }
-
-
+    })
 }
@@ -26,7 +26,6 @@ object Client {

    fun sendRequest(request: HttpRequest, host: String, port: Int, carUid: Int) {
        try {
-            println("sending to $host:$port")
            bootstrap.attr(AttributeKey.valueOf<String>("url"), request.uri())
            bootstrap.attr(AttributeKey.valueOf<Int>("uid"), carUid)
            val ch = bootstrap.connect(host, port).sync().channel()
@@ -1,5 +1,12 @@
 package car.server

+import CodedInputStream
+import CodedOutputStream
+import ConnectionRequest
+import ConnectionResponse
+import InvalidProtocolBufferException
+import RouteDoneRequest
+import RouteDoneResponse
 import connectUrl
 import io.netty.buffer.Unpooled
 import io.netty.channel.ChannelFutureListener
@@ -8,14 +15,8 @@ import io.netty.channel.SimpleChannelInboundHandler
 import io.netty.handler.codec.http.*
 import objects.Environment
 import routeDoneUrl
-import ConnectionRequest
-import ConnectionResponse
 import java.io.ByteArrayInputStream
-import CodedInputStream
-import CodedOutputStream
 import java.io.ByteArrayOutputStream
-import RouteDoneRequest
-import InvalidProtocolBufferException

 /**
 * Created by user on 7/6/16.
@@ -37,13 +38,12 @@ class Handler : SimpleChannelInboundHandler<Any>() {
                    data.mergeFrom(CodedInputStream(ByteArrayInputStream(contentBytes)))
                } catch (e: InvalidProtocolBufferException) {
                    success = false;
+                    ConnectionResponse.BuilderConnectionResponse().setCode(1).setErrorMsg("invalid protobuf request").build().writeTo(CodedOutputStream(answer))
                }
                if (success) {
                    val uid = environment.connectCar(data.ip, data.port)
-
                    ConnectionResponse.BuilderConnectionResponse().setUid(uid).build().writeTo(CodedOutputStream(answer))
                }
-                //todo return connection error
            }
            routeDoneUrl -> {
                val data = RouteDoneRequest.BuilderRouteDoneRequest().build()
@@ -58,11 +58,13 @@ class Handler : SimpleChannelInboundHandler<Any>() {
                        val car = environment.map.get(id)
                        if (car != null) {
                            car.free = true
+                            car.lastAction = System.currentTimeMillis()
+                            RouteDoneResponse.BuilderRouteDoneResponse().setCode(0).setErrorMsg("").build().writeTo(CodedOutputStream(answer))
                        } else {
                            success = false
+                            RouteDoneResponse.BuilderRouteDoneResponse().setCode(2).setErrorMsg("car not found by id").build().writeTo(CodedOutputStream(answer))
                        }
                    })
-                    //todo return connection error
                }
            }
            else -> {
@@ -3,8 +3,6 @@ package car.server
 import io.netty.channel.ChannelInitializer
 import io.netty.channel.ChannelPipeline
 import io.netty.channel.socket.SocketChannel
-import io.netty.handler.codec.http.HttpRequestDecoder
-import io.netty.handler.codec.http.HttpResponseEncoder
 import io.netty.handler.codec.http.HttpServerCodec
 import io.netty.util.concurrent.DefaultEventExecutorGroup
 import io.netty.util.concurrent.EventExecutorGroup
@@ -24,8 +22,6 @@ class Initializer : ChannelInitializer<SocketChannel> {
        val p: ChannelPipeline = channel.pipeline()

        p.addLast(HttpServerCodec())
-//        p.addLast(HttpRequestDecoder())
-//        p.addLast(HttpResponseEncoder())
        p.addLast(group, Handler())
    }
 }
@@ -1,7 +1,5 @@
 package objects

-import car.client.Client
-
 /**
 * Created by user on 7/7/16.
 */
@@ -12,6 +10,7 @@ class Car constructor(uid: Int, host: String, port: Int) {
    val port: Int

    var free: Boolean
+    var lastAction: Long

    var x: Double
    var y: Double
@@ -23,6 +22,7 @@ class Car constructor(uid: Int, host: String, port: Int) {
        this.free = true
        x = 0.toDouble()
        y = 0.toDouble()
+        this.lastAction = System.currentTimeMillis()
    }

    override fun toString(): String {
@@ -1,6 +1,6 @@
 package objects

-import java.util.*
+import java.util.Random

 /**
 * Created by user on 7/7/16.
@@ -19,21 +19,23 @@ class Environment private constructor() {
    }

    @Synchronized
-    fun connectCar(host:String, port:Int):Int {
-        //todo учетка памяти, тут машинки добавляются,но никогда не удаляются. Если включать и выключать одну и ту же машинку, то либо сервер зациклится, либо out of memory
-        //todo в зависимости от того, что кончится раньше, память или айдишники:) Может сделать поток, который мониторит раз в N минут все машинки и дропает те, которые не активны более какого-то времени?
+    fun connectCar(host: String, port: Int): Int {
+        val uid = getNewUid()
+        map.put(uid, Car(uid, host, port))
+        return uid
+    }
+
+    fun getNewUid(): Int {
        var unique = false
        val random = Random()
-        var uid:Int = 0
+        var uid: Int = 0
        while (!unique) {
            uid = random.nextInt(1000000)
            if (map.get(uid) == null) {
                unique = true;
            }
        }
-        map.put(uid, Car(uid, host, port))
-        return uid
+        return uid;
    }

-
 }
@@ -3,8 +3,6 @@ package web.server
 import io.netty.channel.ChannelInitializer
 import io.netty.channel.ChannelPipeline
 import io.netty.channel.socket.SocketChannel
-import io.netty.handler.codec.http.HttpRequestDecoder
-import io.netty.handler.codec.http.HttpResponseEncoder
 import io.netty.handler.codec.http.HttpServerCodec
 import io.netty.util.concurrent.DefaultEventExecutorGroup
 import io.netty.util.concurrent.EventExecutorGroup