diff --git a/server/src/ServerMain.kt b/server/src/ServerMain.kt index cbf141432c1..8beb405e8ea 100644 --- a/server/src/ServerMain.kt +++ b/server/src/ServerMain.kt @@ -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) { - 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) { 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(); + 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 } } - } - - + }) } \ No newline at end of file diff --git a/server/src/car/client/Client.kt b/server/src/car/client/Client.kt index efc0df7c242..acc60f5e6e0 100644 --- a/server/src/car/client/Client.kt +++ b/server/src/car/client/Client.kt @@ -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("url"), request.uri()) bootstrap.attr(AttributeKey.valueOf("uid"), carUid) val ch = bootstrap.connect(host, port).sync().channel() diff --git a/server/src/car/server/Handler.kt b/server/src/car/server/Handler.kt index 36c72316828..4cac677a934 100644 --- a/server/src/car/server/Handler.kt +++ b/server/src/car/server/Handler.kt @@ -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() { 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() { 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 -> { diff --git a/server/src/car/server/Initializer.kt b/server/src/car/server/Initializer.kt index 08d716eaf75..fe79794d319 100644 --- a/server/src/car/server/Initializer.kt +++ b/server/src/car/server/Initializer.kt @@ -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 { val p: ChannelPipeline = channel.pipeline() p.addLast(HttpServerCodec()) -// p.addLast(HttpRequestDecoder()) -// p.addLast(HttpResponseEncoder()) p.addLast(group, Handler()) } } \ No newline at end of file diff --git a/server/src/objects/Car.kt b/server/src/objects/Car.kt index 025bb04ad5f..6400c3e3bde 100644 --- a/server/src/objects/Car.kt +++ b/server/src/objects/Car.kt @@ -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 { diff --git a/server/src/objects/Environment.kt b/server/src/objects/Environment.kt index 69b3dc5b5c6..a64f8c81c19 100644 --- a/server/src/objects/Environment.kt +++ b/server/src/objects/Environment.kt @@ -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; } - } diff --git a/server/src/web/server/Initializer.kt b/server/src/web/server/Initializer.kt index 58f35fac868..79098751d04 100644 --- a/server/src/web/server/Initializer.kt +++ b/server/src/web/server/Initializer.kt @@ -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