iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/sync/PipeInfo.java

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.iotdb.commons.sync;

import org.apache.iotdb.commons.exception.runtime.SerializationRunTimeException;

import org.apache.tsfile.utils.PublicBAOS;
import org.apache.tsfile.utils.ReadWriteIOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.DataOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;

@Deprecated
public abstract class PipeInfo {
  private static final Logger LOGGER = LoggerFactory.getLogger(PipeInfo.class);

  protected String pipeName;
  protected String pipeSinkName;
  protected PipeStatus status;
  protected long createTime;
  protected PipeMessage.PipeMessageType messageType;

  // Only used for serialization
  protected PipeInfo() {
    // Do nothing
  }

  protected PipeInfo(String pipeName, String pipeSinkName, long createTime) {
    this.pipeName = pipeName;
    this.pipeSinkName = pipeSinkName;
    this.createTime = createTime;
    this.status = PipeStatus.STOP;
    this.messageType = PipeMessage.PipeMessageType.NORMAL;
  }

  protected PipeInfo(String pipeName, String pipeSinkName, PipeStatus status, long createTime) {
    this.pipeName = pipeName;
    this.pipeSinkName = pipeSinkName;
    this.createTime = createTime;
    this.status = status;
    this.messageType = PipeMessage.PipeMessageType.NORMAL;
  }

  abstract PipeType getType();

  public String getPipeName() {
    return pipeName;
  }

  public void setPipeName(String pipeName) {
    this.pipeName = pipeName;
  }

  public PipeStatus getStatus() {
    return status;
  }

  public void setStatus(PipeStatus status) {
    this.status = status;
  }

  public void serialize(OutputStream outputStream) throws IOException {
    ReadWriteIOUtils.write((byte) getType().ordinal(), outputStream);
    ReadWriteIOUtils.write(pipeName, outputStream);
    ReadWriteIOUtils.write(pipeSinkName, outputStream);
    ReadWriteIOUtils.write(status.getType(), outputStream);
    ReadWriteIOUtils.write(createTime, outputStream);
    ReadWriteIOUtils.write((byte) messageType.ordinal(), outputStream);
  }

  protected void deserialize(InputStream inputStream) throws IOException {
    pipeName = ReadWriteIOUtils.readString(inputStream);
    pipeSinkName = ReadWriteIOUtils.readString(inputStream);
    status = PipeStatus.getPipeStatus(ReadWriteIOUtils.readByte(inputStream));
    createTime = ReadWriteIOUtils.readLong(inputStream);
    messageType = PipeMessage.PipeMessageType.values()[ReadWriteIOUtils.readByte(inputStream)];
  }

  protected void deserialize(ByteBuffer byteBuffer) {
    pipeName = ReadWriteIOUtils.readString(byteBuffer);
    pipeSinkName = ReadWriteIOUtils.readString(byteBuffer);
    status = PipeStatus.getPipeStatus(ReadWriteIOUtils.readByte(byteBuffer));
    createTime = ReadWriteIOUtils.readLong(byteBuffer);
    messageType = PipeMessage.PipeMessageType.values()[ReadWriteIOUtils.readByte(byteBuffer)];
  }

  public ByteBuffer serializeToByteBuffer() {
    try (PublicBAOS publicBAOS = new PublicBAOS();
        DataOutputStream dataOutputStream = new DataOutputStream(publicBAOS)) {
      serialize(dataOutputStream);
      return ByteBuffer.wrap(publicBAOS.getBuf(), 0, publicBAOS.size());
    } catch (IOException e) {
      LOGGER.error("Unexpected error occurred when serializing PipeInfo.");
      throw new SerializationRunTimeException(e);
    }
  }

  public static PipeInfo deserializePipeInfo(ByteBuffer byteBuffer) {
    PipeType pipeType = PipeType.values()[ReadWriteIOUtils.readByte(byteBuffer)];
    PipeInfo pipeInfo;
    switch (pipeType) {
      case TsFilePipe:
        pipeInfo = new TsFilePipeInfo();
        pipeInfo.deserialize(byteBuffer);
        break;
      case WALPipe:
      default:
        throw new UnsupportedOperationException(
            String.format("Can not recognize PipeType %s.", pipeType.name()));
    }
    return pipeInfo;
  }

  enum PipeType {
    TsFilePipe,
    WALPipe
  }
}