app/src/main/java/net/osdn/gokigen/pkremote/camera/vendor/visionkids/wrapper/playback/MyFtpClient.kt

   1 package net.osdn.gokigen.pkremote.camera.vendor.visionkids.wrapper.playback
   2
   3 import android.util.Log
   4 import net.osdn.gokigen.pkremote.camera.utils.SimpleLogDumper
   5 import java.io.ByteArrayOutputStream
   6 import java.io.DataOutputStream
   7 import java.io.InputStream
   8 import java.lang.Exception
   9 import java.net.InetSocketAddress
  10 import java.net.Socket
  11 import java.util.ArrayDeque
  12 import java.util.Queue
  13
  14 class MyFtpClient(private val callbackReceiver: IFtpServiceCallback, private val isDumpReceiveLog: Boolean = false)
  15 {
  16     private var isStart = false
  17     private var isConnected = false
  18     private var socket: Socket? = null
  19     private var dataOutputStream: DataOutputStream? = null
  20     private var connectedAddress : String? = null
  21     private val commandQueue : Queue<FtpCommand> = ArrayDeque()
  22
  23     private var isStartDataPort = false
  24     private var isConnectedDataPort = false
  25     private var socketDataPort: Socket? = null
  26     private var dataOutputStreamDataPort: DataOutputStream? = null
  27
  28     fun connect(address: String)
  29     {
  30         try
  31         {
  32             connectedAddress = address
  33             Log.v(TAG, "connect to $address")
  34             val thread = Thread {
  35                 try
  36                 {
  37                     val tcpNoDelay = true
  38                     Log.v(TAG, " connect() : $address")
  39                     socket = Socket()
  40                     socket?.reuseAddress = true
  41                     socket?.keepAlive = true
  42                     socket?.tcpNoDelay = true
  43                     if (tcpNoDelay)
  44                     {
  45                         socket?.keepAlive = false
  46                         socket?.setPerformancePreferences(0, 2, 0)
  47                         socket?.oobInline = true
  48                         socket?.reuseAddress = false
  49                         socket?.trafficClass = 0x80
  50                     }
  51                     socket?.connect(InetSocketAddress(address, FTP_CONTROL_PORT), 0)
  52                     dataOutputStream = DataOutputStream(socket?.getOutputStream())
  53                     isConnected = true
  54
  55                     // 接続後の一発目は、自動で読み込んでみる
  56                     val connectCommand = FtpCommand("connect", "connect")
  57                     receiveFromDevice(connectCommand, socket, DATA_POLL_QUEUE_MS, MAX_RETRY_WAIT_COUNT)
  58                     sendCommandMain()
  59                 }
  60                 catch (e: Exception)
  61                 {
  62                     e.printStackTrace()
  63                     callbackReceiver.onReceivedFtpResponse("connect", -1, e.message?:"EXCEPTION")
  64                 }
  65             }
  66             thread.start()
  67         }
  68         catch (e: Exception)
  69         {
  70             e.printStackTrace()
  71         }
  72     }
  73
  74     fun enqueueCommand(command: FtpCommand): Boolean
  75     {
  76         try
  77         {
  78             Log.v(TAG, " Command Enqueue : ${command.command} : ${command.value}")
  79             return commandQueue.offer(command)
  80         }
  81         catch (e: Exception)
  82         {
  83             e.printStackTrace()
  84         }
  85         return (false)
  86     }
  87
  88     fun disconnect()
  89     {
  90         Log.v(TAG, "  ----- DISCONNECT -----")
  91         try
  92         {
  93             // 通信関連のクローズ
  94             closeOutputStream()
  95             closeSocket()
  96             isStart = false
  97             isStartDataPort = false
  98             isConnected = false
  99             isConnectedDataPort = false
 100             commandQueue.clear()
 101             connectedAddress = null
 102         }
 103         catch (e: Exception)
 104         {
 105             e.printStackTrace()
 106         }
 107         System.gc()
 108     }
 109
 110     fun decidePassivePort(response: String)
 111     {
 112         try
 113         {
 114             // データポートの IPアドレスとポート番号を応答データから切り出す
 115             val pickupString = response.substring((response.indexOf("(") + 1), response.indexOf(")"))
 116             val dataStringArray = pickupString.split(",")
 117             val dataPortAddress = dataStringArray[0] + "." + dataStringArray[1] + "." +  dataStringArray[2] + "." +  dataStringArray[3]
 118             val dataPort = dataStringArray[4].toInt() * 256 + dataStringArray[5].toInt()
 119             val passiveAddress = "$dataPortAddress:$dataPort:\r\n"
 120             Log.v(TAG, " - - - - - -  data Port : $passiveAddress ($pickupString  ${dataStringArray.size})")
 121             callbackReceiver.onReceivedFtpResponse("data_port", 0, passiveAddress)
 122         }
 123         catch (e: Exception)
 124         {
 125             e.printStackTrace()
 126         }
 127         sleep(RECEIVE_WAIT_MS)
 128     }
 129
 130     fun openPassivePort(address: String): Boolean
 131     {
 132         var response = true
 133         try
 134         {
 135             // データポートをオープンして受信できるようにする
 136             val accessPoint = address.split(":")
 137             Log.v(TAG, "openPassivePort address:$connectedAddress (or ${accessPoint[0]}) port:${accessPoint[1]}")
 138             val thread = Thread {
 139                 try
 140                 {
 141                     val tcpNoDelay = true
 142                     Log.v(TAG, " connect() : address:${accessPoint[0]} port:${accessPoint[1]}")
 143                     socketDataPort = Socket()
 144                     socketDataPort?.reuseAddress = true
 145                     socketDataPort?.keepAlive = true
 146                     socketDataPort?.tcpNoDelay = true
 147                     if (tcpNoDelay)
 148                     {
 149                         socketDataPort?.keepAlive = false
 150                         socketDataPort?.setPerformancePreferences(0, 2, 0)
 151                         socketDataPort?.oobInline = true
 152                         socketDataPort?.reuseAddress = false
 153                         socketDataPort?.trafficClass = 0x80
 154                     }
 155                     val dataAddress = if (connectedAddress != null) { connectedAddress } else { accessPoint[0] }
 156                     socketDataPort?.connect(InetSocketAddress(dataAddress, accessPoint[1].toInt()), 0)
 157                     dataOutputStreamDataPort = DataOutputStream(socketDataPort?.getOutputStream())
 158                     isConnectedDataPort = true
 159                     receiveDataMain()
 160                 }
 161                 catch (e: Exception)
 162                 {
 163                     e.printStackTrace()
 164                     callbackReceiver.onReceivedFtpResponse("passive_data", -1, e.message?:"EXCEPTION")
 165                 }
 166             }
 167             thread.start()
 168         }
 169         catch (e: Exception)
 170         {
 171             e.printStackTrace()
 172             response = false
 173         }
 174         return (response)
 175     }
 176
 177     private fun closeOutputStream()
 178     {
 179         try
 180         {
 181             dataOutputStream?.close()
 182             dataOutputStreamDataPort?.close()
 183         }
 184         catch (e: Exception)
 185         {
 186             e.printStackTrace()
 187         }
 188         dataOutputStream = null
 189         dataOutputStreamDataPort = null
 190     }
 191
 192     private fun closeSocket()
 193     {
 194         try
 195         {
 196             socket?.close()
 197             socketDataPort?.close()
 198         }
 199         catch (e: Exception)
 200         {
 201             e.printStackTrace()
 202         }
 203         socket = null
 204         socketDataPort = null
 205     }
 206
 207     private fun receiveDataMain()
 208     {
 209         if (isStartDataPort)
 210         {
 211             // すでにコマンドのスレッド動作中なので抜ける
 212             return
 213         }
 214         isStartDataPort = true
 215         Log.v(TAG, " receiveDataMain() : START")
 216         val command = FtpCommand("data", " \r\n")
 217         while (isStartDataPort)
 218         {
 219             try
 220             {
 221                 Log.v(TAG, " --- RECEIVE DATA STANDBY --- ")
 222                 sleep(DATA_POLL_QUEUE_MS)
 223                 receiveFromDevice(command, socketDataPort, DATA_POLL_QUEUE_MS, MAX_RETRY_WAIT_COUNT_DATA)
 224             }
 225             catch (e: Exception)
 226             {
 227                 e.printStackTrace()
 228                 callbackReceiver.onReceivedFtpResponse("receiveDataMain", -1, e.message?:"EXCEPTION")
 229             }
 230         }
 231     }
 232
 233     private fun sendCommandMain()
 234     {
 235         if (isStart)
 236         {
 237             // すでにコマンドのスレッド動作中なので抜ける
 238             return
 239         }
 240         isStart = true
 241         Log.v(TAG, " sendCommandMain() : START")
 242         while (isStart)
 243         {
 244             try
 245             {
 246                 val command = commandQueue.poll()
 247                 if (command != null)
 248                 {
 249                     issueCommand(command)
 250                     sleep(COMMAND_POLL_QUEUE_MS)
 251
 252                     Log.v(TAG, " --- RECEIVE WAIT FOR REPLY --- ")
 253                     receiveFromDevice(command, socket, COMMAND_POLL_QUEUE_MS, MAX_RETRY_WAIT_COUNT)
 254                 }
 255                 sleep(COMMAND_POLL_QUEUE_MS)
 256             }
 257             catch (e: Exception)
 258             {
 259                 e.printStackTrace()
 260                 callbackReceiver.onReceivedFtpResponse("sendCommandMain", -1, e.message?:"EXCEPTION")
 261             }
 262         }
 263     }
 264
 265     private fun receiveFromDevice(command: FtpCommand, targetSocket: Socket?, wait: Int, maxRetry : Int)
 266     {
 267         try
 268         {
 269             val byteArray = ByteArray(PACKET_BUFFER_SIZE)
 270             val inputStream: InputStream? = targetSocket?.getInputStream()
 271             if (inputStream == null)
 272             {
 273                 Log.v(TAG, " InputStream is NULL... RECEIVE ABORTED.")
 274                 callbackReceiver.onReceivedFtpResponse("receiveFromDevice($command)", -1, "InputStream is NULL...")
 275                 return
 276             }
 277
 278             // 初回データが受信バッファにデータが溜まるまで待つ...
 279             var readBytes = waitForReceive(inputStream, wait, maxRetry)
 280             if (readBytes < 0)
 281             {
 282                 // リトライオーバー検出
 283                 Log.v(TAG, "  ----- DETECT RECEIVE RETRY OVER... -----")
 284                 callbackReceiver.onReceivedFtpResponse("receiveFromDevice(${command.command})", -1, "Receive timeout (${COMMAND_POLL_QUEUE_MS * MAX_RETRY_WAIT_COUNT} ms)")
 285             }
 286
 287             // 受信したデータをバッファに突っ込む
 288             var isWriteData = false
 289             //var dataValue = ""
 290             val byteStream = ByteArrayOutputStream()
 291             byteStream.reset()
 292             while (readBytes > 0)
 293             {
 294                 readBytes = inputStream.read(byteArray, 0, PACKET_BUFFER_SIZE)
 295                 if (readBytes <= 0)
 296                 {
 297                     Log.v(TAG," RECEIVED MESSAGE FINISHED ($readBytes)")
 298                     break
 299                 }
 300                 //Log.v(TAG, " :::::::::: [${command.command}] Read Bytes: $readBytes")
 301                 byteStream.write(byteArray, 0, readBytes)
 302                 //dataValue += String(byteArray.copyOfRange(0, readBytes))
 303                 isWriteData = true
 304                 sleep(RECEIVE_WAIT_MS)
 305                 readBytes = inputStream.available()
 306             }
 307             if (isWriteData)
 308             {
 309                 //Log.v(TAG, " >>>>[${command.command}]>>>>>> $dataValue")
 310                 callbackReceiver.onReceivedFtpResponse(command.command, 0, String(byteStream.toByteArray()))
 311                 //callbackReceiver.onReceivedFtpResponse(command.command, 0, dataValue)
 312             }
 313             System.gc()
 314         }
 315         catch (t: Throwable)
 316         {
 317             t.printStackTrace()
 318             callbackReceiver.onReceivedFtpResponse("receiveFromDevice", -1, t.message?:"EXCEPTION")
 319         }
 320     }
 321
 322     private fun issueCommand(command: FtpCommand)
 323     {
 324         try
 325         {
 326             val byteArray = command.value.toByteArray()
 327             if (byteArray.isEmpty())
 328             {
 329                 // メッセージボディがない。終了する
 330                 Log.v(TAG, " SEND BODY IS NOTHING.")
 331                 callbackReceiver.onReceivedFtpResponse(command.command, -1, "SEND COMMAND IS NOTHING.")
 332                 return
 333             }
 334             if (dataOutputStream == null)
 335             {
 336                 Log.v(TAG, " DataOutputStream is null.")
 337                 callbackReceiver.onReceivedFtpResponse(command.command, -1, "DataOutputStream is null.")
 338                 return
 339             }
 340             if (isDumpReceiveLog)
 341             {
 342                 // ログに送信メッセージを出力する
 343                 SimpleLogDumper.dump_bytes("SEND[" + byteArray.size + "] ", byteArray)
 344             }
 345
 346             // (データを)送信
 347             dataOutputStream?.write(byteArray)
 348             dataOutputStream?.flush()
 349         }
 350         catch (e: Exception)
 351         {
 352             e.printStackTrace()
 353             callbackReceiver.onReceivedFtpResponse(command.command, -1, e.message?:"EXCEPTION")
 354         }
 355     }
 356
 357     private fun waitForReceive(inputStream: InputStream, delayMs: Int, maxRetry: Int): Int
 358     {
 359         var retryCount = maxRetry
 360         var isLogOutput = true
 361         var readBytes = 0
 362         try
 363         {
 364             while (readBytes <= 0)
 365             {
 366                 sleep(delayMs)
 367                 readBytes = inputStream.available()
 368                 if (readBytes <= 0)
 369                 {
 370                     if (isLogOutput)
 371                     {
 372                         // Log.v(TAG, "  ----- waitForReceive:: is.available() WAIT... : " + delayMs + "ms")
 373                         isLogOutput = false
 374                     }
 375                     retryCount--
 376                     if (retryCount < 0)
 377                     {
 378                         return (-1)
 379                     }
 380                 }
 381             }
 382         }
 383         catch (e: Exception)
 384         {
 385             e.printStackTrace()
 386         }
 387         return (readBytes)
 388     }
 389
 390     private fun sleep(delayMs: Int)
 391     {
 392         try
 393         {
 394             Thread.sleep(delayMs.toLong())
 395         }
 396         catch (e: Exception)
 397         {
 398             e.printStackTrace()
 399         }
 400     }
 401
 402     companion object {
 403         private val TAG = MyFtpClient::class.java.simpleName
 404         private const val COMMAND_POLL_QUEUE_MS = 15
 405         private const val DATA_POLL_QUEUE_MS = 50
 406         private const val MAX_RETRY_WAIT_COUNT = 20
 407         private const val MAX_RETRY_WAIT_COUNT_DATA = 100
 408         private const val RECEIVE_WAIT_MS = 50
 409         private const val PACKET_BUFFER_SIZE = 8192
 410
 411         private const val FTP_CONTROL_PORT = 21
 412     }
 413 }