opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/cluster/datastore/node/utils/stream/VersionedNormalizedNodeDataInput.java

   1 /*
   2  * Copyright (c) 2019 PANTHEON.tech, s.r.o. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.cluster.datastore.node.utils.stream;
   9
  10 import static java.util.Objects.requireNonNull;
  11
  12 import java.io.DataInput;
  13 import java.io.IOException;
  14
  15 final class VersionedNormalizedNodeDataInput extends ForwardingNormalizedNodeDataInput {
  16     private DataInput input;
  17     private NormalizedNodeDataInput delegate;
  18
  19     VersionedNormalizedNodeDataInput(final DataInput input) {
  20         this.input = requireNonNull(input);
  21     }
  22
  23     @Override
  24     NormalizedNodeDataInput delegate() throws IOException {
  25         if (delegate != null) {
  26             return delegate;
  27         }
  28
  29         final byte marker = input.readByte();
  30         if (marker != TokenTypes.SIGNATURE_MARKER) {
  31             throw defunct("Invalid signature marker: %d", marker);
  32         }
  33
  34         final short version = input.readShort();
  35         final NormalizedNodeDataInput ret;
  36         switch (version) {
  37             case TokenTypes.LITHIUM_VERSION:
  38                 ret = new LithiumNormalizedNodeInputStreamReader(input);
  39                 break;
  40             case TokenTypes.NEON_SR2_VERSION:
  41                 ret = new NeonSR2NormalizedNodeInputStreamReader(input);
  42                 break;
  43             case TokenTypes.SODIUM_SR1_VERSION:
  44                 ret = new SodiumSR1DataInput(input);
  45                 break;
  46             default:
  47                 throw defunct("Unhandled stream version %s", version);
  48         }
  49
  50         setDelegate(ret);
  51         return ret;
  52     }
  53
  54     private InvalidNormalizedNodeStreamException defunct(final String format, final Object... args) {
  55         final InvalidNormalizedNodeStreamException ret = new InvalidNormalizedNodeStreamException(
  56             String.format(format, args));
  57         // Make sure the stream is not touched
  58         setDelegate(new ForwardingNormalizedNodeDataInput() {
  59             @Override
  60             NormalizedNodeDataInput delegate() throws IOException {
  61                 throw new InvalidNormalizedNodeStreamException("Stream is not usable", ret);
  62             }
  63         });
  64         return ret;
  65     }
  66
  67     private void setDelegate(final NormalizedNodeDataInput delegate) {
  68         this.delegate = requireNonNull(delegate);
  69         input = null;
  70     }
  71 }