app/src/main/java/net/osdn/gokigen/a01d/camera/ptpip/wrapper/command/PtpIpCommandPublisher.java

   1 package net.osdn.gokigen.a01d.camera.ptpip.wrapper.command;
   2
   3 import android.util.Log;
   4
   5 import androidx.annotation.NonNull;
   6
   7 import java.io.BufferedReader;
   8 import java.io.ByteArrayOutputStream;
   9 import java.io.DataOutputStream;
  10 import java.io.InputStream;
  11 import java.net.Socket;
  12 import java.util.ArrayDeque;
  13 import java.util.Arrays;
  14 import java.util.Queue;
  15
  16 import static net.osdn.gokigen.a01d.camera.utils.SimpleLogDumper.dump_bytes;
  17
  18 public class PtpIpCommandPublisher implements IPtpIpCommandPublisher, IPtpIpCommunication
  19 {
  20     private static final String TAG = PtpIpCommandPublisher.class.getSimpleName();
  21
  22     private static final int SEQUENCE_START_NUMBER = 1;
  23     private static final int BUFFER_SIZE = 1024 * 1024 + 16;  // 受信バッファは 256kB
  24     private static final int COMMAND_SEND_RECEIVE_DURATION_MS = 5;
  25     private static final int COMMAND_SEND_RECEIVE_DURATION_MAX = 3000;
  26     private static final int COMMAND_POLL_QUEUE_MS = 5;
  27
  28     private final String ipAddress;
  29     private final int portNumber;
  30
  31     private boolean isStart = false;
  32     private boolean isHold = false;
  33     private int holdId = 0;
  34     private Socket socket = null;
  35     private DataOutputStream dos = null;
  36     private BufferedReader bufferedReader = null;
  37     private int sequenceNumber = SEQUENCE_START_NUMBER;
  38     private Queue<IPtpIpCommand> commandQueue;
  39     private Queue<IPtpIpCommand> holdCommandQueue;
  40
  41     public PtpIpCommandPublisher(@NonNull String ip, int portNumber)
  42     {
  43         this.ipAddress = ip;
  44         this.portNumber = portNumber;
  45         this.commandQueue = new ArrayDeque<>();
  46         this.holdCommandQueue = new ArrayDeque<>();
  47         commandQueue.clear();
  48         holdCommandQueue.clear();
  49     }
  50
  51     @Override
  52     public boolean isConnected()
  53     {
  54         return (socket != null);
  55     }
  56
  57     @Override
  58     public boolean connect()
  59     {
  60         try
  61         {
  62             socket = new Socket(ipAddress, portNumber);
  63             return (true);
  64         }
  65         catch (Exception e)
  66         {
  67             e.printStackTrace();
  68             socket = null;
  69         }
  70         return (false);
  71     }
  72
  73     @Override
  74     public void disconnect()
  75     {
  76         // ストリームを全部閉じる
  77         try
  78         {
  79             if (dos != null)
  80             {
  81                 dos.close();
  82             }
  83         }
  84         catch (Exception e)
  85         {
  86             e.printStackTrace();
  87         }
  88         dos = null;
  89
  90         try
  91         {
  92             if (bufferedReader != null)
  93             {
  94                 bufferedReader.close();
  95             }
  96         }
  97         catch (Exception e)
  98         {
  99             e.printStackTrace();
 100         }
 101         bufferedReader = null;
 102
 103         try
 104         {
 105             if (socket != null)
 106             {
 107                 socket.close();
 108             }
 109         }
 110         catch (Exception e)
 111         {
 112             e.printStackTrace();
 113         }
 114         socket = null;
 115         sequenceNumber = SEQUENCE_START_NUMBER;
 116         isStart = false;
 117         commandQueue.clear();
 118         System.gc();
 119     }
 120
 121     @Override
 122     public void start()
 123     {
 124         if (isStart)
 125         {
 126             // すでにコマンドのスレッド動作中なので抜ける
 127             return;
 128         }
 129         isStart = true;
 130
 131         Thread thread = new Thread(new Runnable()
 132         {
 133             @Override
 134             public void run()
 135             {
 136                 try
 137                 {
 138                     dos = new DataOutputStream(socket.getOutputStream());
 139                     while (isStart)
 140                     {
 141                         try
 142                         {
 143                             IPtpIpCommand command = commandQueue.poll();
 144                             if (command != null)
 145                             {
 146                                 issueCommand(command);
 147                             }
 148                             Thread.sleep(COMMAND_POLL_QUEUE_MS);
 149                             // Log.v(TAG, " QUEUE SIZE : " + commandQueue.size());
 150                         }
 151                         catch (Exception e)
 152                         {
 153                             e.printStackTrace();
 154                         }
 155                     }
 156                 }
 157                 catch (Exception e)
 158                 {
 159                     Log.v(TAG, "<<<<< IP : " + ipAddress + " port : " + portNumber + " >>>>>");
 160                     e.printStackTrace();
 161                 }
 162             }
 163         });
 164         try
 165         {
 166             thread.start();
 167         }
 168         catch (Exception e)
 169         {
 170             e.printStackTrace();
 171         }
 172     }
 173
 174     @Override
 175     public void stop()
 176     {
 177         isStart = false;
 178         commandQueue.clear();
 179     }
 180
 181     @Override
 182     public boolean enqueueCommand(@NonNull IPtpIpCommand command)
 183     {
 184         try
 185         {
 186             if (isHold) {
 187                 if (holdId == command.getHoldId()) {
 188                     if (command.isRelease()) {
 189                         // コマンドをキューに積んだ後、リリースする
 190                         boolean ret = commandQueue.offer(command);
 191                         isHold = false;
 192
 193                         //  溜まっているキューを積みなおす
 194                         while (holdCommandQueue.size() != 0) {
 195                             IPtpIpCommand queuedCommand = holdCommandQueue.poll();
 196                             commandQueue.offer(queuedCommand);
 197                             if ((queuedCommand != null)&&(queuedCommand.isHold()))
 198                             {
 199                                 // 特定シーケンスに入った場合は、そこで積みなおすのをやめる
 200                                 isHold = true;
 201                                 holdId = queuedCommand.getHoldId();
 202                                 break;
 203                             }
 204                         }
 205                         return (ret);
 206                     }
 207                     return (commandQueue.offer(command));
 208                 } else {
 209                     // 特定シーケンスではなかったので HOLD
 210                     return (holdCommandQueue.offer(command));
 211                 }
 212             }
 213             if (command.isHold())
 214             {
 215                 isHold = true;
 216                 holdId = command.getHoldId();
 217             }
 218             //Log.v(TAG, "Enqueue : "  + command.getId());
 219             return (commandQueue.offer(command));
 220         }
 221         catch (Exception e)
 222         {
 223             e.printStackTrace();
 224         }
 225         return (false);
 226     }
 227
 228     @Override
 229     public boolean flushHoldQueue()
 230     {
 231         Log.v(TAG, "  flushHoldQueue()");
 232         holdCommandQueue.clear();
 233         System.gc();
 234         return (true);
 235     }
 236
 237     private void issueCommand(@NonNull IPtpIpCommand command)
 238     {
 239         try
 240         {
 241             boolean retry_over = true;
 242             while (retry_over)
 243             {
 244                 //Log.v(TAG, "issueCommand : " + command.getId());
 245                 byte[] commandBody = command.commandBody();
 246                 if (commandBody != null)
 247                 {
 248                     // コマンドボディが入っていた場合には、コマンド送信（入っていない場合は受信待ち）
 249                     send_to_camera(command.dumpLog(), commandBody, command.useSequenceNumber(), command.embeddedSequenceNumberIndex());
 250                     byte[] commandBody2 = command.commandBody2();
 251                     if (commandBody2 != null)
 252                     {
 253                         // コマンドボディの２つめが入っていた場合には、コマンドを連続送信する
 254                         send_to_camera(command.dumpLog(), commandBody2, command.useSequenceNumber(), command.embeddedSequenceNumberIndex2());
 255                     }
 256                     byte[] commandBody3 = command.commandBody3();
 257                     if (commandBody3 != null)
 258                     {
 259                         // コマンドボディの３つめが入っていた場合には、コマンドを連続送信する
 260                         send_to_camera(command.dumpLog(), commandBody3, command.useSequenceNumber(), command.embeddedSequenceNumberIndex3());
 261                     }
 262                     if (command.isIncrementSeqNumber())
 263                     {
 264                         // シーケンス番号を更新する
 265                         sequenceNumber++;
 266                     }
 267                 }
 268                 retry_over = receive_from_camera(command);
 269                 if ((retry_over)&&(commandBody != null))
 270                 {
 271                     if (!command.isRetrySend())
 272                     {
 273                         //  コマンドを再送信しない場合はここで抜ける
 274                         break;
 275                     }
 276                     // 再送信...のために、シーケンス番号を戻す...
 277                     sequenceNumber--;
 278                 }
 279             }
 280         }
 281         catch (Exception e)
 282         {
 283             e.printStackTrace();
 284         }
 285     }
 286
 287     /**
 288      *    カメラにコマンドを送信する（メイン部分）
 289      *
 290      */
 291     private void send_to_camera(boolean isDumpReceiveLog, byte[] byte_array, boolean useSequenceNumber, int embeddedSequenceIndex)
 292     {
 293         try
 294         {
 295             if (dos == null)
 296             {
 297                 Log.v(TAG, " DataOutputStream is null.");
 298                 return;
 299             }
 300
 301             //dos = new DataOutputStream(socket.getOutputStream());  // ここにいたらいけない？
 302
 303             // メッセージボディを加工： 最初に４バイトのレングス長をつける
 304             byte[] sendData = new byte[byte_array.length + 4];
 305
 306             sendData[0] = (byte) (byte_array.length + 4);
 307             sendData[1] = 0x00;
 308             sendData[2] = 0x00;
 309             sendData[3] = 0x00;
 310             System.arraycopy(byte_array,0,sendData,4, byte_array.length);
 311
 312             if (useSequenceNumber)
 313             {
 314                 // Sequence Number を反映させる
 315                 sendData[embeddedSequenceIndex] = (byte) ((0x000000ff & sequenceNumber));
 316                 sendData[embeddedSequenceIndex + 1] = (byte) (((0x0000ff00 & sequenceNumber) >>> 8) & 0x000000ff);
 317                 sendData[embeddedSequenceIndex + 2] = (byte) (((0x00ff0000 & sequenceNumber) >>> 16) & 0x000000ff);
 318                 sendData[embeddedSequenceIndex + 3] = (byte) (((0xff000000 & sequenceNumber) >>> 24) & 0x000000ff);
 319                 if (isDumpReceiveLog)
 320                 {
 321                     Log.v(TAG, "----- SEQ No. : " + sequenceNumber + " -----");
 322                 }
 323             }
 324
 325             if (isDumpReceiveLog)
 326             {
 327                 // ログに送信メッセージを出力する
 328                 dump_bytes("SEND[" + sendData.length + "] ", sendData);
 329             }
 330
 331             // (データを)送信
 332             dos.write(sendData);
 333             dos.flush();
 334         }
 335         catch (Exception e)
 336         {
 337             e.printStackTrace();
 338         }
 339     }
 340
 341     private void sleep(int delayMs)
 342     {
 343         try
 344         {
 345             Thread.sleep(delayMs);
 346         }
 347         catch (Exception e)
 348         {
 349             e.printStackTrace();
 350         }
 351     }
 352
 353
 354     /**
 355      *    カメラからにコマンドの結果を受信する（メイン部分）
 356      *
 357      */
 358     private boolean receive_from_camera(@NonNull IPtpIpCommand command)
 359     {
 360         IPtpIpCommandCallback callback = command.responseCallback();
 361         int delayMs = command.receiveDelayMs();
 362         if ((delayMs < 0)||(delayMs > COMMAND_SEND_RECEIVE_DURATION_MAX))
 363         {
 364             delayMs = COMMAND_SEND_RECEIVE_DURATION_MS;
 365         }
 366         if ((callback != null)&&(callback.isReceiveMulti()))
 367         {
 368             // 受信したら逐次「受信したよ」と応答するパターン
 369             return (receive_multi(command, delayMs));
 370         }
 371         //  受信した後、すべてをまとめて「受信したよ」と応答するパターン
 372         return (receive_single(command, delayMs));
 373     }
 374
 375     private boolean receive_single(@NonNull IPtpIpCommand command, int delayMs)
 376     {
 377         boolean isDumpReceiveLog = command.dumpLog();
 378         int id = command.getId();
 379         IPtpIpCommandCallback callback = command.responseCallback();
 380         try
 381         {
 382             int receive_message_buffer_size = BUFFER_SIZE;
 383             byte[] byte_array = new byte[receive_message_buffer_size];
 384             InputStream is = socket.getInputStream();
 385             if (is == null)
 386             {
 387                 Log.v(TAG, " InputStream is NULL... RECEIVE ABORTED.");
 388                 return (false);
 389             }
 390
 391             // 初回データが受信バッファにデータが溜まるまで待つ...
 392             int read_bytes = waitForReceive(is, delayMs);
 393             if (read_bytes < 0)
 394             {
 395                 // リトライオーバー...
 396                 Log.v(TAG, " RECEIVE : RETRY OVER...");
 397                 return (true);
 398             }
 399
 400             // 受信したデータをバッファに突っ込む
 401             ByteArrayOutputStream byteStream = new ByteArrayOutputStream();
 402             while (read_bytes > 0)
 403             {
 404                 read_bytes = is.read(byte_array, 0, receive_message_buffer_size);
 405                 if (read_bytes <= 0)
 406                 {
 407                     Log.v(TAG, " RECEIVED MESSAGE FINISHED (" + read_bytes + ")");
 408                     break;
 409                 }
 410                 byteStream.write(byte_array, 0, read_bytes);
 411                 sleep(delayMs);
 412                 read_bytes = is.available();
 413             }
 414             ByteArrayOutputStream outputStream = cutHeader(byteStream);
 415             receivedAllMessage(isDumpReceiveLog, id, outputStream.toByteArray(), callback);
 416             System.gc();
 417         }
 418         catch (Throwable e)
 419         {
 420             e.printStackTrace();
 421             System.gc();
 422         }
 423         return (false);
 424     }
 425
 426     private void receivedAllMessage(boolean isDumpReceiveLog, int id, byte[] body, IPtpIpCommandCallback callback)
 427     {
 428         Log.v(TAG, "receivedAllMessage() : " + ((body == null) ? 0 : body.length) + " bytes.");
 429         if ((isDumpReceiveLog)&&(body != null))
 430         {
 431             // ログに受信メッセージを出力する
 432             dump_bytes("RECV[" + body.length + "] ", body);
 433         }
 434         if (callback != null)
 435         {
 436             callback.receivedMessage(id, body);
 437         }
 438     }
 439
 440     private boolean receive_multi(@NonNull IPtpIpCommand command, int delayMs)
 441     {
 442         int estimatedSize = command.estimatedReceiveDataSize();
 443         int id = command.getId();
 444         IPtpIpCommandCallback callback = command.responseCallback();
 445
 446         try
 447         {
 448             Log.v(TAG, " ===== receive_multi() =====");
 449             int receive_message_buffer_size = BUFFER_SIZE;
 450             byte[] byte_array = new byte[receive_message_buffer_size];
 451             InputStream is = socket.getInputStream();
 452             if (is == null)
 453             {
 454                 Log.v(TAG, " InputStream is NULL... RECEIVE ABORTED.");
 455                 return (false);
 456             }
 457
 458             // 初回データが受信バッファにデータが溜まるまで待つ...
 459             int read_bytes = waitForReceive(is, delayMs);
 460             if (read_bytes < 0)
 461             {
 462                 // リトライオーバー...
 463                 Log.v(TAG, " RECEIVE : RETRY OVER......");
 464                 return (true);
 465             }
 466
 467             // 初回データの読み込み
 468             read_bytes = is.read(byte_array, 0, receive_message_buffer_size);
 469             int target_length = parseDataLength(byte_array, read_bytes);
 470             int received_length = read_bytes;
 471
 472             //  一時的な処理
 473             if (callback != null)
 474             {
 475                 Log.v(TAG, "  --- 1st CALL : read_bytes : "+ read_bytes + "(" + received_length + ") : total_length : " + target_length + "  buffer SIZE : " + byte_array.length);
 476                 callback.onReceiveProgress(received_length, target_length, Arrays.copyOfRange(byte_array, 0, received_length));
 477             }
 478
 479             do
 480             {
 481                 sleep(delayMs);
 482                 read_bytes = is.available();
 483                 if (read_bytes == 0)
 484                 {
 485                     Log.v(TAG, " WAIT is.available() ... " + received_length + " < " + estimatedSize);
 486                 }
 487             } while ((read_bytes == 0)&&(estimatedSize > 0)&&(received_length < estimatedSize));
 488             while (read_bytes > 0)
 489             {
 490                 read_bytes = is.read(byte_array, 0, receive_message_buffer_size);
 491                 if (read_bytes <= 0)
 492                 {
 493                     Log.v(TAG, "  RECEIVED MESSAGE FINISHED (" + read_bytes + ")");
 494                     break;
 495                 }
 496                 received_length = received_length + read_bytes;
 497
 498                 //  一時的な処理
 499                 if (callback != null)
 500                 {
 501                     //Log.v(TAG, "  --- CALL : read_bytes : "+ read_bytes + " total_read : " + received_length + " : total_length : " + target_length + "  buffer SIZE : " + byte_array.length);
 502                     callback.onReceiveProgress(received_length, target_length, Arrays.copyOfRange(byte_array, 0, read_bytes));
 503                 }
 504                 //byteStream.write(byte_array, 0, read_bytes);
 505
 506                 do
 507                 {
 508                     sleep(delayMs);
 509                     read_bytes = is.available();
 510                     //Log.v(TAG, "  is.available() read_bytes : " + read_bytes + " " + received_length + " < " + estimatedSize);
 511                 } while ((read_bytes == 0)&&(estimatedSize > 0)&&(received_length < estimatedSize));
 512             }
 513             //ByteArrayOutputStream outputStream = cutHeader(byteStream);
 514             //receivedMessage(isDumpReceiveLog, id, outputStream.toByteArray(), callback);
 515
 516             //  終了報告...一時的？
 517             if (callback != null)
 518             {
 519                 Log.v(TAG, "  --- receive_multi : receivedMessage() : " + id + "  (" + read_bytes + ") [" + estimatedSize + "] " + receive_message_buffer_size + " (" + received_length + ")");
 520                 callback.receivedMessage(id, null);
 521             }
 522             System.gc();
 523         }
 524         catch (Throwable e)
 525         {
 526             e.printStackTrace();
 527             System.gc();
 528         }
 529         return (false);
 530     }
 531
 532     private int parseDataLength(byte[] byte_array, int read_bytes)
 533     {
 534         int lenlen = 0;
 535         int packetType = 0;
 536         try
 537         {
 538             if ((read_bytes > 20)&&((int) byte_array[4] == 0x09))
 539             {
 540                 lenlen = ((((int) byte_array[15]) & 0xff) << 24) + ((((int) byte_array[14]) & 0xff) << 16) + ((((int) byte_array[13]) & 0xff) << 8) + (((int) byte_array[12]) & 0xff);
 541                 packetType = (((int)byte_array[16]) & 0xff);
 542             }
 543             Log.v(TAG, " --- parseDataLength() length: " + lenlen + " TYPE: " + packetType + " read_bytes: " + read_bytes);
 544         }
 545         catch (Exception e)
 546         {
 547             e.printStackTrace();
 548         }
 549         return (lenlen);
 550     }
 551
 552     private ByteArrayOutputStream cutHeader(ByteArrayOutputStream receivedBuffer)
 553     {
 554         try
 555         {
 556             byte[] byte_array = receivedBuffer.toByteArray();
 557             int limit = byte_array.length;
 558             int lenlen = 0;
 559             int len = ((((int) byte_array[3]) & 0xff) << 24) + ((((int) byte_array[2]) & 0xff) << 16) + ((((int) byte_array[1]) & 0xff) << 8) + (((int) byte_array[0]) & 0xff);
 560             int packetType = (((int) byte_array[4]) & 0xff);
 561             if ((limit == len)||(limit < 16384))
 562             {
 563                 // 応答は１つしか入っていない。もしくは受信データサイズが16kBの場合は、そのまま返す。
 564                 return (receivedBuffer);
 565             }
 566             if (packetType == 0x09)
 567             {
 568                 lenlen = ((((int) byte_array[15]) & 0xff) << 24) + ((((int) byte_array[14]) & 0xff) << 16) + ((((int) byte_array[13]) & 0xff) << 8) + (((int) byte_array[12]) & 0xff);
 569                 packetType = (((int) byte_array[16]) & 0xff);
 570             }
 571             Log.v(TAG, " ---  RECEIVED MESSAGE : " + len + " bytes (BUFFER: " + byte_array.length + " bytes)" + " length : " + lenlen + " TYPE : " + packetType + " --- ");
 572             if (lenlen == 0)
 573             {
 574                 // データとしては変なので、なにもしない
 575                 return (receivedBuffer);
 576             }
 577             ByteArrayOutputStream outputStream =  new ByteArrayOutputStream();
 578             //outputStream.write(byte_array, 0, 20);  //
 579             int position = 20;  // ヘッダ込の先頭
 580             while (position < limit)
 581             {
 582                 lenlen = ((((int) byte_array[position + 3]) & 0xff) << 24) + ((((int) byte_array[position + 2]) & 0xff) << 16) + ((((int) byte_array[position + 1]) & 0xff) << 8) + (((int) byte_array[position]) & 0xff);
 583                 packetType = (((int) byte_array[position + 4]) & 0xff);
 584 /*
 585                 if (packetType != 0x0a)
 586                 {
 587                     Log.v(TAG, " <><><> PACKET TYPE : " + packetType + " LENGTH : " + lenlen);
 588                 }
 589 */
 590                 int copyByte = ((lenlen - 12) > (limit - (position + 12))) ? (limit - (position + 12)) : (lenlen - 12);
 591                 outputStream.write(byte_array, (position + 12), copyByte);
 592                 position = position + lenlen;
 593             }
 594             return (outputStream);
 595         }
 596         catch (Throwable e)
 597         {
 598             e.printStackTrace();
 599             System.gc();
 600         }
 601         return (receivedBuffer);
 602     }
 603
 604     private int waitForReceive(InputStream is, int delayMs)
 605     {
 606         boolean isLogOutput = true;
 607         int retry_count = 50;
 608         int read_bytes = 0;
 609         try
 610         {
 611             while (read_bytes <= 0)
 612             {
 613                 sleep(delayMs);
 614                 read_bytes = is.available();
 615                 if (read_bytes == 0)
 616                 {
 617                     if (isLogOutput)
 618                     {
 619                         Log.v(TAG, " is.available() WAIT... ");
 620                         isLogOutput = false;
 621                     }
 622                     retry_count--;
 623                     if (retry_count < 0)
 624                     {
 625                         return (-1);
 626                     }
 627                 }
 628             }
 629         }
 630         catch (Exception e)
 631         {
 632             e.printStackTrace();
 633         }
 634         return (read_bytes);
 635     }
 636 }