src/main/java/com/att/nsa/cambria/backends/memory/MemoryConsumerFactory.java

   1 /*******************************************************************************
   2  *  ============LICENSE_START=======================================================
   3  *  org.onap.dmaap
   4  *  ================================================================================
   5  *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
   6  *  ================================================================================
   7  *  Licensed under the Apache License, Version 2.0 (the "License");
   8  *  you may not use this file except in compliance with the License.
   9  *  You may obtain a copy of the License at
  10  *        http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  *  Unless required by applicable law or agreed to in writing, software
  13  *  distributed under the License is distributed on an "AS IS" BASIS,
  14  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  *  See the License for the specific language governing permissions and
  16  *  limitations under the License.
  17  *  ============LICENSE_END=========================================================
  18  *
  19  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  20  *
  21  *******************************************************************************/
  22 package com.att.nsa.cambria.backends.memory;
  23
  24 import java.util.ArrayList;
  25 import java.util.Collection;
  26
  27 import com.att.nsa.cambria.backends.Consumer;
  28 import com.att.nsa.cambria.backends.ConsumerFactory;
  29 /**
  30  *
  31  * @author author
  32  *
  33  */
  34 public class MemoryConsumerFactory implements ConsumerFactory
  35 {
  36         /**
  37          *
  38          * Initializing constructor
  39          * @param q
  40          */
  41         public MemoryConsumerFactory ( MemoryQueue q )
  42         {
  43                 fQueue = q;
  44         }
  45
  46         /**
  47          *
  48          * @param topic
  49          * @param consumerGroupId
  50          * @param clientId
  51          * @param timeoutMs
  52          * @return Consumer
  53          */
  54         @Override
  55         public Consumer getConsumerFor ( String topic, String consumerGroupId, String clientId, int timeoutMs )
  56         {
  57                 return new MemoryConsumer ( topic, consumerGroupId );
  58         }
  59
  60         private final MemoryQueue fQueue;
  61
  62         /**
  63          *
  64          * Define nested inner class
  65          *
  66          */
  67         private class MemoryConsumer implements Consumer
  68         {
  69                 /**
  70                  *
  71                  * Initializing MemoryConsumer constructor
  72                  * @param topic
  73                  * @param consumer
  74                  *
  75                  */
  76                 public MemoryConsumer ( String topic, String consumer )
  77                 {
  78                         fTopic = topic;
  79                         fConsumer = consumer;
  80                         fCreateMs = System.currentTimeMillis ();
  81                         fLastAccessMs = fCreateMs;
  82                 }
  83
  84                 @Override
  85                 /**
  86                  *
  87                  * return consumer details
  88                  */
  89                 public Message nextMessage ()
  90                 {
  91                         return fQueue.get ( fTopic, fConsumer );
  92                 }
  93
  94                 private final String fTopic;
  95                 private final String fConsumer;
  96                 private final long fCreateMs;
  97                 private long fLastAccessMs;
  98
  99                 @Override
 100                 public void close() {
 101                         //Nothing to close/clean up.
 102                 }
 103                 /**
 104                  *
 105                  */
 106                 public void commitOffsets()
 107                 {
 108                         // ignoring this aspect
 109                 }
 110                 /**
 111                  * get offset
 112                  */
 113                 public long getOffset()
 114                 {
 115                         return 0;
 116                 }
 117
 118                 @Override
 119                 /**
 120                  * get consumer topic name
 121                  */
 122                 public String getName ()
 123                 {
 124                         return fTopic + "/" + fConsumer;
 125                 }
 126
 127                 @Override
 128                 public long getCreateTimeMs ()
 129                 {
 130                         return fCreateMs;
 131                 }
 132
 133                 @Override
 134                 public long getLastAccessMs ()
 135                 {
 136                         return fLastAccessMs;
 137                 }
 138         }
 139
 140         @Override
 141         public void destroyConsumer(String topic, String consumerGroupId,
 142                         String clientId) {
 143                 //No cache for memory consumers, so NOOP
 144         }
 145
 146         @Override
 147         public void dropCache ()
 148         {
 149                 // nothing to do - there's no cache here
 150         }
 151
 152         @Override
 153         /**
 154          * @return ArrayList<MemoryConsumer>
 155          */
 156         public Collection<? extends Consumer> getConsumers ()
 157         {
 158                 return new ArrayList<MemoryConsumer> ();
 159         }
 160 }