How to use QueueConsumer class of mock.async package

Best Karate code snippet using mock.async.QueueConsumer

Source:EventPublisherTest.java

1package mil.tron.commonapi.pubsub;2import com.google.common.collect.Lists;3import mil.tron.commonapi.entity.AppClientUser;4import mil.tron.commonapi.entity.Person;5import mil.tron.commonapi.entity.Privilege;6import mil.tron.commonapi.entity.pubsub.Subscriber;7import mil.tron.commonapi.entity.pubsub.events.EventType;8import mil.tron.commonapi.pubsub.messages.PersonChangedMessage;9import mil.tron.commonapi.pubsub.messages.PubSubMessage;10import mil.tron.commonapi.service.pubsub.SubscriberService;11import mil.tron.commonapi.service.pubsub.log.EventRequestService;12import mil.tron.commonapi.service.utility.IstioHeaderUtils;13import org.junit.jupiter.api.AfterEach;14import org.junit.jupiter.api.BeforeEach;15import org.junit.jupiter.api.Test;16import org.junit.jupiter.api.extension.ExtendWith;17import org.mockito.InjectMocks;18import org.mockito.Mockito;19import org.springframework.beans.factory.annotation.Autowired;20import org.springframework.boot.test.context.SpringBootTest;21import org.springframework.boot.test.mock.mockito.MockBean;22import org.springframework.test.context.TestPropertySource;23import org.springframework.test.context.junit.jupiter.SpringExtension;24import org.springframework.util.StringUtils;25import org.springframework.web.client.RestClientException;26import org.springframework.web.client.RestTemplate;27import java.io.ByteArrayOutputStream;28import java.io.PrintStream;29import java.net.URI;30import java.util.Set;31import java.util.UUID;32import static org.junit.jupiter.api.Assertions.*;33@ExtendWith(SpringExtension.class)34@SpringBootTest(classes=EventPublisher.class, properties = { "webhook-queue-max-size=5" })  // set limit to reasonable value35@TestPropertySource(locations = "classpath:application-test.properties")36public class EventPublisherTest {37    @Autowired38    @InjectMocks39    private EventPublisher publisher;40    @MockBean41    private SubscriberService subService;42    43    @MockBean44    private EventRequestService eventRequestService;45    @MockBean(name="eventSender")46    private RestTemplate publisherSender;47    private Subscriber subscriber, subscriber2;48    private PrintStream originalSystemOut = System.out;49    private final ByteArrayOutputStream outputStreamCaptor = new ByteArrayOutputStream();50    // mimic a real-formatted x-forwarded-client-cert header field that would be from a POST'r to the common api51    //  we'll see thru the code that this POSTr's XFCC header is from puckboard.52    //  The logic should see change is from puckboard and thus wont send push notification back to puckboard53    //  since they're the ones that originated the change54    private final String uri = "By=spiffe://cluster.local/ns/tron-common-api/sa/default;Hash=blah;Subject=\"\";URI=spiffe://cluster.local/ns/tron-puckboard/sa/default";55    @BeforeEach56    void setupMockSubscriber() {57        AppClientUser user1 = AppClientUser.builder()58                .name("Test")59                .clusterUrl("http://some.svc.cluster.local")60                .privileges(Set.of(Privilege.builder().id(1L).name("PERSON_READ").build()))61                .build();62        AppClientUser user2 = AppClientUser.builder()63                .name("Test")64                .clusterUrl("http://puckboard-api-service.tron-puckboard.svc.cluster.local")65                .privileges(Set.of(Privilege.builder().id(1L).name("PERSON_READ").build()))66                .build();67        subscriber = Subscriber.builder()68                .id(UUID.randomUUID())69                .appClientUser(user1)70                .subscribedEvent(EventType.PERSON_CHANGE)71                .subscriberAddress("/api/changed")72                .build();73        subscriber2 = Subscriber.builder()74                .id(UUID.randomUUID())75                .appClientUser(user2)76                .subscribedEvent(EventType.PERSON_CHANGE)77                // mimic real-formatted puckboard cluster URI as a subscriber78                .subscriberAddress("/puckboard-api/v1")79                .build();80        originalSystemOut = System.out;81        System.setOut(new PrintStream(outputStreamCaptor));82    }83    @AfterEach84    public void tearDown() throws Exception {85        System.setOut(originalSystemOut);86    }87    @Test88    void testAsyncPublish() {89        Mockito.when(subService.getSubscriptionsByEventType(Mockito.any(EventType.class)))90                .thenReturn(Lists.newArrayList(subscriber, subscriber2));91        Mockito.when(publisherSender.postForLocation(Mockito.anyString(), Mockito.any(PubSubMessage.class)))92                .thenReturn(URI.create(subscriber.getSubscriberAddress()));93        PersonChangedMessage message = new PersonChangedMessage();94        message.addPersonId(new Person().getId());95        publisher.publishEvent(message, uri);96        publisher.queueConsumer();  // manually tick the consumer97        // wait for publishEvent Async function, its a mocked function, so 1sec it more than enough but needed to avoid98        // a race condition on the logging output getting captured99        try { Thread.sleep(1000); } catch (InterruptedException e) {}100        assertTrue(outputStreamCaptor.toString().contains("[PUBLISH BROADCAST]"));101        assertFalse(outputStreamCaptor.toString().contains("[PUBLISH ERROR]"));102        // make sure we have only one broadcast push sent out of 2 subscribers (puckboard subscriber is omitted in this case)103        assertEquals(1, StringUtils.countOccurrencesOf(outputStreamCaptor.toString(), "[PUBLISH BROADCAST]"));104    }105    @Test106    void testAsyncPublishWithMalformedXFCCHeader() {107        // a malformed XFCC should just be ignored, and we just blast out the message to all subscribers108        Mockito.when(subService.getSubscriptionsByEventType(Mockito.any(EventType.class)))109                .thenReturn(Lists.newArrayList(subscriber, subscriber2));110        Mockito.when(publisherSender.postForLocation(Mockito.anyString(), Mockito.any(PubSubMessage.class)))111                .thenReturn(URI.create(subscriber.getSubscriberAddress()));112        PersonChangedMessage message = new PersonChangedMessage();113        message.addPersonId(new Person().getId());114        publisher.publishEvent(message, "");115        publisher.queueConsumer();  // manually tick the consumer116        // wait for publishEvent Async function, its a mocked function, so 1sec it more than enough but needed to avoid117        // a race condition on the logging output getting captured118        try { Thread.sleep(1000); } catch (InterruptedException e) {}119        assertTrue(outputStreamCaptor.toString().contains("[PUBLISH BROADCAST]"));120        assertFalse(outputStreamCaptor.toString().contains("[PUBLISH ERROR]"));121        assertEquals(2, StringUtils.countOccurrencesOf(outputStreamCaptor.toString(), "[PUBLISH BROADCAST]"));122    }123    @Test124    void testAsyncPublishFails() {125        Mockito.when(subService.getSubscriptionsByEventType(Mockito.any(EventType.class)))126                .thenReturn(Lists.newArrayList(subscriber));127        Mockito.when(128                publisherSender.postForLocation(Mockito.anyString(), Mockito.any()))129                .thenThrow(new RestClientException("Exception"));130        PersonChangedMessage message = new PersonChangedMessage();131        message.addPersonId(new Person().getId());132        publisher.publishEvent(message, uri);133        publisher.queueConsumer();  // manually tick the consumer134        // wait for publishEvent Async function, its a mocked function, so 1sec it more than enough but needed to avoid135        // a race condition on the logging output getting capture136        try { Thread.sleep(1000); } catch (InterruptedException e) {}137        assertTrue(outputStreamCaptor.toString().contains("[PUBLISH ERROR]"));138    }139    @Test140    void testQueueMaxedOut() {141        PersonChangedMessage message = new PersonChangedMessage();142        message.addPersonId(new Person().getId());143        // queue up 6 messages which exceeds our 5 limit setup for testing144        publisher.publishEvent(message, uri);145        publisher.publishEvent(message, uri);146        publisher.publishEvent(message, uri);147        publisher.publishEvent(message, uri);148        publisher.publishEvent(message, uri);149        publisher.publishEvent(message, uri);150        assertTrue(outputStreamCaptor.toString().contains("MAX QUEUE SIZE reached"));151        // dequeue everything152        publisher.queueConsumer();153        publisher.queueConsumer();154        publisher.queueConsumer();155        publisher.queueConsumer();156        publisher.queueConsumer();157    }158    @Test159    void testExtractNameSpaceFromURI() {160        assertEquals("tron-puckboard", IstioHeaderUtils.extractSubscriberNamespace("http://puckboard-api-service.tron-puckboard.svc.cluster.local/puckboard-api/v1"));161        assertEquals("", IstioHeaderUtils.extractSubscriberNamespace("http://cvc.cluster.local/puckboard-api/v1"));162        assertEquals("", IstioHeaderUtils.extractSubscriberNamespace("http://svc.cluster.local/puckboard-api/v1"));163        assertEquals("3000", IstioHeaderUtils.extractSubscriberNamespace("http://localhost:3000"));164        assertEquals("", IstioHeaderUtils.extractSubscriberNamespace(null));165        assertEquals("tron-puckboard", publisher.extractNamespace("By=spiffe://cluster.local/ns/tron-common-api/sa/default;Hash=blah;Subject=\"\";URI=spiffe://cluster.local/ns/tron-puckboard/sa/default"));166        assertEquals("3000", publisher.extractNamespace("By=spiffe://cluster.local/ns/tron-common-api/sa/default;Hash=blah;Subject=\"\";URI=spiffe://localhost:3000"));167        assertNull(publisher.extractNamespace(""));168    }169}...

Source:QueueConsumer.java

...10import javax.jms.Session;11import javax.jms.TextMessage;12import org.slf4j.Logger;13import org.slf4j.LoggerFactory;14public class QueueConsumer {15    private static final Logger logger = LoggerFactory.getLogger(QueueConsumer.class);16    public static final String QUEUE_NAME = "MOCK.ASYNC";17    private final Connection connection;18    private final MessageConsumer consumer;19    private final Session session;20    // in more complex tests or for re-usability, this field and append() /21    // collect() / clear() methods can be in a separate / static class22    private final List messages = new ArrayList();23    public synchronized void append(Object message) {24        messages.add(message);25        if (condition.test(message)) {26            logger.debug("condition met, will signal completion");27            future.complete(Boolean.TRUE);28        } else {29            logger.debug("condition not met, will continue waiting");30        }31    }32    public synchronized List collect() {33        return messages;34    }35    36    private CompletableFuture future = new CompletableFuture();37    private Predicate condition = o -> true; // just a default38    39    // note how you can pass data in from the test for very dynamic checks40    public List waitUntilCount(int count) {        41        condition = o -> messages.size() == count;42        try {43            future.get(5000, TimeUnit.MILLISECONDS);44        } catch (Exception e) {45            logger.error("wait timed out: {}", e + "");46        }47        return messages;48    }49    public QueueConsumer() {50        this.connection = QueueUtils.getConnection();51        try {52            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);53            Destination destination = session.createQueue(QUEUE_NAME);54            consumer = session.createConsumer(destination);55            consumer.setMessageListener(message -> {56                TextMessage tm = (TextMessage) message;57                try {58                    // this is where we "collect" messages for assertions later59                    append(tm.getText());60                } catch (Exception e) {61                    throw new RuntimeException(e);62                }63            });...

QueueConsumer

Using AI Code Generation

1import mock.async.QueueConsumer;2import java.util.Queue;3import java.util.LinkedList;4import java.util.concurrent.ExecutorService;5import java.util.concurrent.Executors;6import java.util.concurrent.TimeUnit;7import java.util.concurrent.atomic.AtomicInteger;8import java.util.concurrent.atomic.AtomicLong;9public class QueueConsumerTest {10    public static void main(String[] args) throws InterruptedException {11        final Queue<String> queue = new LinkedList<String>();12        final AtomicInteger counter = new AtomicInteger();13        final AtomicLong total = new AtomicLong();14        final long start = System.currentTimeMillis();15        final ExecutorService executor = Executors.newFixedThreadPool(4);16        for (int i = 0; i < 4; i++) {17            executor.execute(new QueueConsumer<String>(queue) {18                protected void consume(String item) {19                    total.addAndGet(item.length());20                    counter.incrementAndGet();21                }22            });23        }24        for (int i = 0; i < 100000; i++) {25            queue.add("Hello World");26        }27        executor.shutdown();28        executor.awaitTermination(1, TimeUnit.DAYS);29        System.out.println("Total time: " + (System.currentTimeMillis() - start));30        System.out.println("Total items: " + counter.get());31        System.out.println("Total length: " + total.get());32    }33}34import mock.async.BlockingQueueConsumer;35import java.util.Queue;36import java.util.LinkedList;37import java.util.concurrent.ExecutorService;38import java.util.concurrent.Executors;39import java.util.concurrent.TimeUnit;40import java.util.concurrent.atomic.AtomicInteger;41import java.util.concurrent.atomic.AtomicLong;42public class BlockingQueueConsumerTest {43    public static void main(String[] args) throws InterruptedException {44        final Queue<String> queue = new LinkedList<String>();45        final AtomicInteger counter = new AtomicInteger();46        final AtomicLong total = new AtomicLong();47        final long start = System.currentTimeMillis();48        final ExecutorService executor = Executors.newFixedThreadPool(4);49        for (int i = 0; i < 4; i++) {50            executor.execute(new BlockingQueueConsumer<String>(queue) {51                protected void consume(String item) {52                    total.addAndGet(item.length());53                    counter.incrementAndGet();54                }55            });56        }57        for (int i = 0; i < 100000; i++) {58            queue.add("Hello World");59        }

QueueConsumer

Using AI Code Generation

1import mock.async.*;2public class 4 {3    public static void main(String[] args) {4        QueueConsumer qc = new QueueConsumer();5        qc.consume();6    }7}8import mock.async.*;9public class 5 {10    public static void main(String[] args) {11        QueueConsumer qc = new QueueConsumer();12        qc.consume();13    }14}15import mock.async.*;16public class 6 {17    public static void main(String[] args) {18        QueueConsumer qc = new QueueConsumer();19        qc.consume();20    }21}22import mock.async.*;23public class 7 {24    public static void main(String[] args) {25        QueueConsumer qc = new QueueConsumer();26        qc.consume();27    }28}29import mock.async.*;30public class 8 {31    public static void main(String[] args) {32        QueueConsumer qc = new QueueConsumer();33        qc.consume();34    }35}36import mock.async.*;37public class 9 {38    public static void main(String[] args) {39        QueueConsumer qc = new QueueConsumer();40        qc.consume();41    }42}43import mock.async.*;44public class 10 {45    public static void main(String[] args) {46        QueueConsumer qc = new QueueConsumer();47        qc.consume();48    }49}50import mock.async.*;51public class 11 {52    public static void main(String[] args) {53        QueueConsumer qc = new QueueConsumer();54        qc.consume();55    }56}57import mock.async.*;58public class 12 {59    public static void main(String[] args) {60        QueueConsumer qc = new QueueConsumer();61        qc.consume();62    }63}64import mock.async.*;65public class 13 {

QueueConsumer

Using AI Code Generation

1import java.util.concurrent.BlockingQueue;2import java.util.concurrent.LinkedBlockingQueue;3import java.util.concurrent.TimeUnit;4import java.util.concurrent.atomic.AtomicInteger;5import mock.async.QueueConsumer;6public class Main {7	public static void main(String[] args) {8		BlockingQueue<Integer> queue = new LinkedBlockingQueue<Integer>();9		AtomicInteger counter = new AtomicInteger();10		QueueConsumer<Integer> consumer = new QueueConsumer<Integer>(queue, new Consumer(counter));11		consumer.start();12		try {13			for (int i = 0; i < 100; i++) {14				queue.put(i);15			}16			queue.put(-1);17		} catch (InterruptedException e) {18			e.printStackTrace();19		}20		consumer.stop();21		System.out.println("Total: " + counter.get());22	}23}24class Consumer implements Runnable {25	private AtomicInteger counter;26	public Consumer(AtomicInteger counter) {27		this.counter = counter;28	}29	public void run() {30		System.out.println("Consumer started");31		while (true) {32			try {33				int i = counter.incrementAndGet();34				System.out.println("Consuming " + i);35				TimeUnit.MILLISECONDS.sleep(100);36			} catch (InterruptedException e) {37				System.out.println("Consumer stopped");38				return;39			}40		}41	}42}43package mock.async;44import java.util.concurrent.BlockingQueue;45import java.util.concurrent.TimeUnit;46public class QueueConsumer<T> {47	private BlockingQueue<T> queue;48	private volatile boolean stop = false;49	private Thread thread;50	private Runnable consumer;51	public QueueConsumer(BlockingQueue<T> queue, Runnable consumer) {52		this.queue = queue;53		this.consumer = consumer;54	}55	public void start() {56		thread = new Thread(new Runnable() {57			public void run() {58				while (!stop) {59					try {60						T item = queue.poll(100, TimeUnit.MILLISECONDS);61						if (item != null) {62							consumer.run();63						}64					} catch (InterruptedException e) {65						e.printStackTrace();66					}67				}68			}69		});70		thread.start();71	}72	public void stop() {73		stop = true;74		try {75			thread.join();76		} catch (InterruptedException e) {77			e.printStackTrace();78		}79	}80}

QueueConsumer

Using AI Code Generation

1import mock.async.QueueConsumer;2public class Main {3  public static void main(String[] args) {4    QueueConsumer qc = new QueueConsumer();5    qc.consume();6  }7}8import mock.async.QueueConsumer;9public class Main {10  public static void main(String[] args) {11    QueueConsumer qc = new QueueConsumer();12    qc.consume();13  }14}15import mock.async.QueueConsumer;16public class Main {17  public static void main(String[] args) {18    QueueConsumer qc = new QueueConsumer();19    qc.consume();20  }21}22import mock.async.QueueConsumer;23public class Main {24  public static void main(String[] args) {25    QueueConsumer qc = new QueueConsumer();26    qc.consume();27  }28}29import mock.async.QueueConsumer;30public class Main {31  public static void main(String[] args) {32    QueueConsumer qc = new QueueConsumer();33    qc.consume();34  }35}36import mock.async.QueueConsumer;37public class Main {38  public static void main(String[] args) {39    QueueConsumer qc = new QueueConsumer();40    qc.consume();41  }42}43import mock.async.QueueConsumer;44public class Main {45  public static void main(String[] args) {46    QueueConsumer qc = new QueueConsumer();47    qc.consume();48  }49}50import mock.async.QueueConsumer;51public class Main {52  public static void main(String[] args) {53    QueueConsumer qc = new QueueConsumer();54    qc.consume();55  }56}57import mock.async.QueueConsumer;58public class Main {59  public static void main(String[] args) {60    QueueConsumer qc = new QueueConsumer();61    qc.consume();62  }63}64import mock.async

QueueConsumer

Using AI Code Generation

1import mock.async.QueueConsumer;2import java.util.concurrent.ArrayBlockingQueue;3import java.util.concurrent.BlockingQueue;4public class 4 {5    public static void main(String[] args) {6        BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(10);7        QueueConsumer consumer = new QueueConsumer(queue);8        Thread thread = new Thread(consumer);9        thread.start();10        for(int i=0; i<100; i++) {11            try {12                queue.put(i);13            } catch (InterruptedException e) {14                e.printStackTrace();15            }16        }17    }18}

QueueConsumer

Using AI Code Generation

1import mock.async.*;2class QueueConsumerTest {3    public static void main(String[] args) {4        QueueConsumer q = new QueueConsumer();5        q.start();6        q.stop();7    }8}9package mock.async;10public class QueueConsumer {11    public void start() {12        System.out.println("start");13    }14    public void stop() {15        System.out.println("stop");16    }17}

QueueConsumer

Using AI Code Generation

1import java.util.*;2{3	public static void main(String[] args)4	{5		QueueConsumer qc = new QueueConsumer();6		qc.start();7	}8}9import java.util.*;10import java.io.*;11import java.net.*;12import java.util.concurrent.*;13{14	private final int MAX_SIZE = 50;15	private final int MIN_SIZE = 0;16	private int size;17	private Queue<String> queue;18	private int count;19	public QueueConsumer()20	{21		size = 0;22		queue = new LinkedList<String>();23		count = 0;24	}25	public void addElement(String element)26	{27		if(size == MAX_SIZE)28		{29			System.out.println("Queue is full");30		}31		{32			queue.add(element);33			size++;34		}35	}36	public void removeElement()37	{38		if(size == MIN_SIZE)39		{40			System.out.println("Queue is empty");41		}42		{43			queue.remove();44			size--;45		}46	}47	public void display()48	{49		if(size == MIN_SIZE)50		{51			System.out.println("Queue is empty");52		}53		{54			System.out.println(queue);55		}56	}57	public void start()58	{59		while(true)60		{61			{62				if(size != MIN_SIZE)63				{64					System.out.println("Consumer is consuming element");65					removeElement();66					count++;67					System.out.println("Element consumed");68					System.out.println("No. of elements consumed: " + count);69					System.out.println("No. of elements in queue: " + size);70					System.out.println("Queue: " + queue);71					Thread.sleep(5000);72				}73				{74					System.out.println("No element in queue");75					System.out.println("Producer is producing element");76					addElement("Element");77					System.out.println("Element produced");78					System.out.println("No. of elements in queue: "

QueueConsumer

Using AI Code Generation

1import mock.async.*;2import java.util.*;3import java.io.*;4import java.lang.*;5{6public static void main(String args[])7{8{9QueueConsumer qc = new QueueConsumer();10qc.receive();11}12catch(Exception e)13{14System.out.println(e);15}16}17}18package mock.async;19import java.util.*;20import java.io.*;21import java.lang.*;22{23public void receive() throws Exception24{25Scanner sc = new Scanner(System.in);26System.out.println("enter the number of elements in queue");27int n = sc.nextInt();28Queue queue = new LinkedList();29for(int i=0;i<n;i++)30{31System.out.println("enter the element");32int element = sc.nextInt();33queue.add(element);34}35System.out.println("enter the number of elements to be removed");36int m = sc.nextInt();37for(int i=0;i<m;i++)38{39System.out.println("enter the element to be removed");40int element = sc.nextInt();41queue.remove(element);42}43System.out.println("the elements in the queue are");44for(Object element:queue)45{46System.out.println(element);47}48}49}

QueueConsumer

Using AI Code Generation

1import mock.async.QueueConsumer;2public class 4 {3    public static void main(String[] args) {4        QueueConsumer consumer = new QueueConsumer();5        consumer.consume();6    }7}

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.