001/* 002 * (C) Copyright 2017 Nuxeo SA (http://nuxeo.com/) and others. 003 * 004 * Licensed under the Apache License, Version 2.0 (the "License"); 005 * you may not use this file except in compliance with the License. 006 * You may obtain a copy of the License at 007 * 008 * http://www.apache.org/licenses/LICENSE-2.0 009 * 010 * Unless required by applicable law or agreed to in writing, software 011 * distributed under the License is distributed on an "AS IS" BASIS, 012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 013 * See the License for the specific language governing permissions and 014 * limitations under the License. 015 * 016 * Contributors: 017 * bdelbosc 018 */ 019package org.nuxeo.lib.stream.pattern.producer; 020 021import static org.nuxeo.lib.stream.codec.NoCodec.NO_CODEC; 022 023import java.util.List; 024import java.util.concurrent.Callable; 025 026import org.apache.commons.logging.Log; 027import org.apache.commons.logging.LogFactory; 028import org.nuxeo.lib.stream.codec.Codec; 029import org.nuxeo.lib.stream.log.LogAppender; 030import org.nuxeo.lib.stream.log.LogManager; 031import org.nuxeo.lib.stream.log.Name; 032import org.nuxeo.lib.stream.pattern.Message; 033import org.nuxeo.lib.stream.pattern.consumer.internals.AbstractCallablePool; 034import org.nuxeo.lib.stream.pattern.producer.internals.ProducerRunner; 035 036/** 037 * A Pool of ProducerRunner 038 * 039 * @since 9.1 040 */ 041public class ProducerPool<M extends Message> extends AbstractCallablePool<ProducerStatus> { 042 private static final Log log = LogFactory.getLog(ProducerPool.class); 043 044 protected final LogManager manager; 045 046 protected final String logName; 047 048 protected final Codec<M> codec; 049 050 protected final ProducerFactory<M> factory; 051 052 /** 053 * @deprecated since 11.1, due to serialization issue with java 11, use 054 * {@link #ProducerPool(String, LogManager, Codec, ProducerFactory, short)} which allows to give a 055 * {@link org.nuxeo.lib.stream.codec.Codec codec} to {@link LogAppender appender}. 056 */ 057 @Deprecated 058 @SuppressWarnings("unchecked") 059 public ProducerPool(String logName, LogManager manager, ProducerFactory<M> factory, short nbThreads) { 060 this(logName, manager, NO_CODEC, factory, nbThreads); 061 } 062 063 public ProducerPool(String logName, LogManager manager, Codec<M> codec, ProducerFactory<M> factory, 064 short nbThreads) { 065 super(nbThreads); 066 this.logName = logName; 067 this.manager = manager; 068 this.codec = codec; 069 this.factory = factory; 070 } 071 072 @Override 073 protected ProducerStatus getErrorStatus() { 074 return new ProducerStatus(0, 0, 0, 0, true); 075 } 076 077 @Override 078 protected Callable<ProducerStatus> getCallable(int i) { 079 return new ProducerRunner<>(factory, manager.getAppender(Name.ofUrn(logName), codec), i); 080 } 081 082 @Override 083 protected String getThreadPrefix() { 084 return "Nuxeo-Producer"; 085 } 086 087 @Override 088 protected void afterCall(List<ProducerStatus> ret) { 089 ret.forEach(log::info); 090 log.warn(ProducerStatus.toString(ret)); 091 } 092 093}