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