package tmp;
import com.google.common.base.Charsets;
import io.netty.buffer.ByteBuf;
import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.client.api.DigestType;
import org.apache.bookkeeper.client.api.LedgerEntries;
import org.apache.bookkeeper.client.api.LedgerEntry;
import org.apache.bookkeeper.client.api.ReadHandle;
import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.apache.pulsar.common.protocol.Commands;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
 * @function
 * @date 2021/8/10 16:10
 */
public class BKMain {
    private static final Logger log = LoggerFactory.getLogger(BKMain.class);
    public static void main(String[] args) throws Exception {
        BKMain main = new BKMain();
        main.run();
    }
    void run() throws Exception {
        load01();
    }
    void load01() throws Exception {
        // zookeeper地址
        // bk的信息保存在zk的/ledger路径
        String connectionString = "x.x.x.x:2181";
        BookKeeper bkClient = new BookKeeper(connectionString);
        // 通过zk上/managed-ledgers/tenant_c/ns1/persistent/storeV2获取topic storeV2的ledgerId
        long ledgerId = 56990;
        // 设置加密方式和密码（同pulsar一样）
        ReadHandle handle = bkClient.newOpenLedgerOp().withRecovery(false).withLedgerId(ledgerId)
                .withDigestType(DigestType.CRC32C).withPassword("".getBytes(Charsets.UTF_8)).execute().get();
        // 获取上一次被添加确认的entryId。则这个entryId之前的数据对外是可见的
        long id = handle.readLastAddConfirmed();
        log.info("id=" + id);
        long firstEntry = 0;
        long lastEntry = id;
        LedgerEntries entries = handle.read(firstEntry, lastEntry);
        if (entries != null) {
            StringBuilder sb = new StringBuilder("\n");
            for (LedgerEntry entry : entries) {
                ByteBuf buf = entry.getEntryBuffer();
                // 解析每一条消息（因为我们本身没有采用批量的方式发送消息，没有压缩，没有加密，而且指定的schema也是string，所以解析部分也很简单）
                Commands.skipBrokerEntryMetadataIfExist(buf);
                Commands.skipChecksumIfPresent(buf);
                sb.setLength(0);
                sb.append("ledgerId=").append(entry.getLedgerId());
                sb.append(", entryId=").append(entry.getEntryId());
                sb.append(", dataLen=").append(entry.getLength() + buf.readableBytes());
                MessageMetadata msgMetadata = Commands.parseMessageMetadata(buf);
                sb.append(", numMessages=").append(msgMetadata.getNumMessagesInBatch());
                sb.append(", publishTime=").append(msgMetadata.getPublishTime());
                sb.append(", sequenceId=").append(msgMetadata.getSequenceId());
                // 如果有些字段不是必须的，则需要先判断是否有该字段，否则解析会报错。
                sb.append(", numChunks=").append(msgMetadata.hasNumChunksFromMsg() ? msgMetadata.getNumChunksFromMsg() : 0);
                sb.append(", replicateFrom=").append(msgMetadata.hasReplicatedFrom() ? msgMetadata.getReplicatedFrom() : "none");
                byte[] dataBytes = new byte[buf.readableBytes()];
                buf.readBytes(dataBytes);
                sb.append(", data=").append(new String(dataBytes));
                log.info(sb.toString());
            }
            entries.close();
        }
        handle.close();
        bkClient.close();
    }
}

@Override
protected Message<T> internalReceive(int timeout, TimeUnit unit) throws PulsarClientException {
    Message<T> message;
    try {
        // 先从缓存队列中获取，如果获取不到则直接返回。
        message = incomingMessages.poll(timeout, unit);
        if (message == null) {
            return null;
        }
        // 暂时不返回消息，而是给broker发送通知，表示自己可以接收更多的消息了。
        messageProcessed(message);
        return beforeConsume(message);
    } catch (InterruptedException e) {
        State state = getState();
        if (state != State.Closing && state != State.Closed) {
            stats.incrementNumReceiveFailed();
            throw PulsarClientException.unwrap(e);
        } else {
            return null;
        }
    }
}
@Override
protected synchronized void messageProcessed(Message<?> msg) {
    ClientCnx currentCnx = cnx();
    ClientCnx msgCnx = ((MessageImpl<?>) msg).getCnx();
    lastDequeuedMessageId = msg.getMessageId();
    if (msgCnx != currentCnx) {
        // The processed message did belong to the old queue that was cleared after reconnection.
    } else {
        increaseAvailablePermits(currentCnx);
        stats.updateNumMsgsReceived(msg);
        trackMessage(msg);
    }
    decreaseIncomingMessageSize(msg);
}
void increaseAvailablePermits(ClientCnx currentCnx) {
    increaseAvailablePermits(currentCnx, 1);
}
protected void increaseAvailablePermits(ClientCnx currentCnx, int delta) {
    int available = AVAILABLE_PERMITS_UPDATER.addAndGet(this, delta);
    while (available >= receiverQueueRefillThreshold && !paused) {
        // 确保能发送一次通知
        if (AVAILABLE_PERMITS_UPDATER.compareAndSet(this, available, 0)) {
            sendFlowPermitsToBroker(currentCnx, available);
            break;
        } else {
            available = AVAILABLE_PERMITS_UPDATER.get(this);
        }
    }
}
/**
 * send the flow command to have the broker start pushing messages
 * 发送flow通知给broker，异步获取更多消息。
 */
private void sendFlowPermitsToBroker(ClientCnx cnx, int numMessages) {
    if (cnx != null && numMessages > 0) {
        if (log.isDebugEnabled()) {
            log.debug("[{}] [{}] Adding {} additional permits", topic, subscription, numMessages);
        }
        if (log.isDebugEnabled()) {
            cnx.ctx().writeAndFlush(Commands.newFlow(consumerId, numMessages))
                .addListener(writeFuture -> {
                    if (!writeFuture.isSuccess()) {
                        log.debug("Consumer {} failed to send {} permits to broker: {}", consumerId, numMessages,
                                  writeFuture.cause().getMessage());
                    } else {
                        log.debug("Consumer {} sent {} permits to broker", consumerId, numMessages);
                    }
                });
        } else {
            cnx.ctx().writeAndFlush(Commands.newFlow(consumerId, numMessages), cnx.ctx().voidPromise());
        }
    }
}

public void testZKMetaData() throws Exception {
    String path = "/managed-ledgers/tenant_c/ns1/persistent/topic_name";
    byte[] bytes = getValue(path);
    if (bytes == null) {
        System.out.println("bytes is null");
        return;
    }
    MLDataFormats.ManagedLedgerInfo info = MLDataFormats.ManagedLedgerInfo.parseFrom(bytes);
    StringBuilder sb = new StringBuilder();
    for (MLDataFormats.ManagedLedgerInfo.LedgerInfo ls : info.getLedgerInfoList()) {
        sb.setLength(0);
        sb.append("ledgerId=").append(ls.getLedgerId());
        sb.append(", entries=").append(ls.getEntries());
        sb.append(", size=").append(ls.getSize());
        sb.append(", timestamp=").append(ls.getTimestamp());
        System.out.println("debug vv " + sb);
    }
    System.out.println("done");
}
public void testZKMetaDataCursorInfo() throws Exception {
    String path = "/managed-ledgers/tenant_c/ns1/persistent/topic_name/subscription_name_0";
    path = "/managed-ledgers/tenant_c/ns1/persistent/topic_name/subscription_name_1";
    byte[] bytes = getValue(path);
    if (bytes == null) {
        System.out.println("bytes is null");
        return;
    }
    MLDataFormats.ManagedCursorInfo info = MLDataFormats.ManagedCursorInfo.parseFrom(bytes);
    StringBuilder sb = new StringBuilder();
    sb.setLength(0);
    sb.append("markDeleteLedgerId=").append(info.getMarkDeleteLedgerId());
    sb.append(", markDeleteEntryId=").append(info.getMarkDeleteEntryId());
    sb.append(", lastActive=").append(info.getLastActive());
    System.out.println("debug vv " + sb);
    System.out.println("done");
}
byte[] getValue(String path) throws Exception {
    CountDownLatch latch = new CountDownLatch(1);
    ZooKeeper zoo = new ZooKeeper("x.x.x.x:2181", 10000, new Watcher() {
        @Override
        public void process(WatchedEvent event) {
            if (event.getState() == Event.KeeperState.SyncConnected) {
                latch.countDown();
            }
            System.out.println("state " + event.getState());
        }
    });
    latch.await();
    byte[] bytes = zoo.getData(path, null, null);
    zoo.close();
    return bytes;
}

package encrypt;
import org.apache.pulsar.client.api.CryptoKeyReader;
import org.apache.pulsar.client.api.EncryptionKeyInfo;
import java.io.File;
import java.nio.file.Files;
import java.util.Map;
/**
 * @function
 * @date 2021/7/27 9:45
 */
public class RawFileKeyReader implements CryptoKeyReader {
    String publicKeyFile = "F:\\download\\test_ecdsa_pubkey.pem";
    String privateKeyFile = "F:\\download\\test_ecdsa_privkey.pem";
    @Override
    public EncryptionKeyInfo getPublicKey(String keyName, Map<String, String> metadata) {
        EncryptionKeyInfo info = new EncryptionKeyInfo();
        try {
            System.out.println("getPublicKey");
            info.setKey(Files.readAllBytes(new File(publicKeyFile).toPath()));
        } catch (Exception e) {
            e.printStackTrace();
        }
        return info;
    }
    @Override
    public EncryptionKeyInfo getPrivateKey(String keyName, Map<String, String> metadata) {
        EncryptionKeyInfo info = new EncryptionKeyInfo();
        System.out.println("getPrivateKey");
        try {
            info.setKey(Files.readAllBytes(new File(privateKeyFile).toPath()));
        } catch (Exception e) {
            e.printStackTrace();
        }
        return info;
    }
}

package encrypt;
import auth.client.VVAuthentication;
import org.apache.pulsar.client.api.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.TimeUnit;
/**
 
 * @function
 * @date 2021/7/27 8:47
 */
public class EncryptTest {
    private static final Logger log = LoggerFactory.getLogger(EncryptTest.class);
    public static void main(String[] args) throws Exception {
        EncryptTest main = new EncryptTest();
        main.run();
    }
    String topic = "persistent://tenant_c/ns1/topic1";
    void run() throws Exception {
        PulsarClient client = PulsarClient.builder()
                .authentication(new VVAuthentication())
                .serviceUrl("pulsar://172.20.140.11:6650")
                .build();
        produce(client);
        consume(client);
        client.close();
    }
    void produce(PulsarClient client) throws Exception {
        Producer p = client.newProducer()
                .topic(topic)
                .producerName("pro_encrypt")
                // 设置AES密钥，如果不设置则不会对消息加密
                .addEncryptionKey("vv_01")
                // 设置从哪获取公钥/私钥
                .cryptoKeyReader(new RawFileKeyReader())
                .create();
        MessageId id = p.newMessage(Schema.STRING).key("key-1").value("hello world").send();
        p.flush();
        log.info("send " + id);
        p.close();
    }
    void consume(PulsarClient client) throws Exception {
        Consumer c = client.newConsumer()
                .subscriptionName("con_encrypt")
                .topic(topic)
                .cryptoKeyReader(new RawFileKeyReader())
                .subscribe();
        while (true) {
            Message msg = c.receive(2, TimeUnit.SECONDS);
            if (msg == null) {
                break;
            }
            c.acknowledge(msg);
            log.info("receive " + msg.getKey() + ", " + new String(msg.getData()));
        }
        c.close();
    }
}

package ack;
import auth.client.VVAuthentication;
import org.apache.pulsar.client.api.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
/**
 * @function
 * @date 2021/8/4 16:23
 */
public class RedeliverTest {
    private static final Logger log = LoggerFactory.getLogger(RedeliverTest.class);
    public static void main(String[] args) throws Exception {
        RedeliverTest main = new RedeliverTest();
        main.run();
    }
    String topic = "persistent://tenant_vv/ns1/redeliver_test";
    void run() throws Exception {
        CountDownLatch latch = new CountDownLatch(1);
        new Thread(() -> consumer(true, latch)).start();
        
        // 确保第一个消费者先收到消息
        latch.await();
        new Thread(() -> consumer(false, latch)).start();
        produce();
        log.info("done");
    }
    void produce() throws Exception {
        try (PulsarClient client = newClient()) {
            Producer p = client.newProducer()
                    .topic(topic)
                    .create();
            for (int i=0; i<2; i++) {
                p.newMessage(Schema.STRING).value("data_" + i).key("sub mode "+ i).sendAsync();
            }
            p.flush();
            p.close();
        } catch (Exception e) {
            log.error("", e);
        }
    }
    void consumer(boolean replay, CountDownLatch latch)  {
        try (PulsarClient client = newClient()) {
            String tag = "replay_mode_" + replay;
            Consumer c = client.newConsumer(Schema.STRING)
                    .topic(topic)
                    // 设置消息处理失败后，2秒后再此收到消息。
                    .negativeAckRedeliveryDelay(2, TimeUnit.SECONDS)
                    .subscriptionType(SubscriptionType.Shared)
                    .subscriptionName("sub_name")
                    .subscribe();
            latch.countDown();
            while (true) {
                Message<String> msg = c.receive(10, TimeUnit.SECONDS);
                if (msg == null) {
                    break;
                }
                if (replay) {
                    c.negativeAcknowledge(msg);
                } else {
                    c.acknowledge(msg);
                }
                log.info(tag + " receive " + msg.getValue());
            }
            c.close();
        } catch (Exception e) {
            log.error("", e);
        }
    }
    PulsarClient newClient() throws Exception {
        PulsarClient client = PulsarClient.builder()
                .serviceUrl("pulsar://x.x.x.x:6650")
                .authentication(new VVAuthentication())
                .build();
        return client;
    }
}

@Override
public void negativeAcknowledge(MessageId messageId) {
    negativeAcksTracker.add(messageId);
    // Ensure the message is not redelivered for ack-timeout, since we did receive an "ack"
    unAckedMessageTracker.remove(messageId);
}

@Override
public void redeliverUnacknowledgedMessages(Set<MessageId> messageIds) {
    if (messageIds.isEmpty()) {
        return;
    }
    checkArgument(messageIds.stream().findFirst().get() instanceof MessageIdImpl);
    if (conf.getSubscriptionType() != SubscriptionType.Shared
        && conf.getSubscriptionType() != SubscriptionType.Key_Shared) {
        // We cannot redeliver single messages if subscription type is not Shared
        redeliverUnacknowledgedMessages();
        return;
    }
    ClientCnx cnx = cnx();
    if (isConnected() && cnx.getRemoteEndpointProtocolVersion() >= ProtocolVersion.v2.getValue()) {
        int messagesFromQueue = removeExpiredMessagesFromQueue(messageIds);
        Iterable<List<MessageIdImpl>> batches = Iterables.partition(
            messageIds.stream()
            .map(messageId -> (MessageIdImpl)messageId)
            .collect(Collectors.toSet()), MAX_REDELIVER_UNACKNOWLEDGED);
        batches.forEach(ids -> {
            getRedeliveryMessageIdData(ids).thenAccept(messageIdData -> {
                if (!messageIdData.isEmpty()) {
                    ByteBuf cmd = Commands.newRedeliverUnacknowledgedMessages(consumerId, messageIdData);
                    cnx.ctx().writeAndFlush(cmd, cnx.ctx().voidPromise());
                }
            });
        });
        if (messagesFromQueue > 0) {
            increaseAvailablePermits(cnx, messagesFromQueue);
        }
        if (log.isDebugEnabled()) {
            log.debug("[{}] [{}] [{}] Redeliver unacked messages and increase {} permits", subscription, topic,
                      consumerName, messagesFromQueue);
        }
        return;
    }
    if (cnx == null || (getState() == State.Connecting)) {
        log.warn("[{}] Client Connection needs to be established for redelivery of unacknowledged messages", this);
    } else {
        log.warn("[{}] Reconnecting the client to redeliver the messages.", this);
        cnx.ctx().close();
    }
}

public synchronized void add(MessageId messageId) {
    if (messageId instanceof BatchMessageIdImpl) {
        BatchMessageIdImpl batchMessageId = (BatchMessageIdImpl) messageId;
        messageId = new MessageIdImpl(batchMessageId.getLedgerId(), batchMessageId.getEntryId(),
                                      batchMessageId.getPartitionIndex());
    }
    if (nackedMessages == null) {
        nackedMessages = new HashMap<>();
    }
    nackedMessages.put(messageId, System.nanoTime() + nackDelayNanos);
    if (this.timeout == null) {
        // Schedule a task and group all the redeliveries for same period. Leave a small buffer to allow for
        // nack immediately following the current one will be batched into the same redeliver request.
        this.timeout = timer.newTimeout(this::triggerRedelivery, timerIntervalNanos, TimeUnit.NANOSECONDS);
    }
}

private synchronized void triggerRedelivery(Timeout t) {
    if (nackedMessages.isEmpty()) {
        this.timeout = null;
        return;
    }
    // Group all the nacked messages into one single re-delivery request
    Set<MessageId> messagesToRedeliver = new HashSet<>();
    long now = System.nanoTime();
    nackedMessages.forEach((msgId, timestamp) -> {
        if (timestamp < now) {
            addChunkedMessageIdsAndRemoveFromSequnceMap(msgId, messagesToRedeliver, this.consumer);
            messagesToRedeliver.add(msgId);
        }
    });
    messagesToRedeliver.forEach(nackedMessages::remove);
    consumer.onNegativeAcksSend(messagesToRedeliver);
    // 使用netty发送消息给broker
    consumer.redeliverUnacknowledgedMessages(messagesToRedeliver);
    this.timeout = timer.newTimeout(this::triggerRedelivery, timerIntervalNanos, TimeUnit.NANOSECONDS);
}

@Override
protected void handleRedeliverUnacknowledged(CommandRedeliverUnacknowledgedMessages redeliver) {
    checkArgument(state == State.Connected);
    if (log.isDebugEnabled()) {
        log.debug("[{}] Received Resend Command from consumer {} ", remoteAddress, redeliver.getConsumerId());
    }
    CompletableFuture<Consumer> consumerFuture = consumers.get(redeliver.getConsumerId());
    if (consumerFuture != null && consumerFuture.isDone() && !consumerFuture.isCompletedExceptionally()) {
        Consumer consumer = consumerFuture.getNow(null);
        if (redeliver.getMessageIdsCount() > 0 && Subscription.isIndividualAckMode(consumer.subType())) {
            consumer.redeliverUnacknowledgedMessages(redeliver.getMessageIdsList());
        } else {
            consumer.redeliverUnacknowledgedMessages();
        }
    }
}

public void redeliverUnacknowledgedMessages(List<MessageIdData> messageIds) {
    int totalRedeliveryMessages = 0;
    List<PositionImpl> pendingPositions = Lists.newArrayList();
    for (MessageIdData msg : messageIds) {
        PositionImpl position = PositionImpl.get(msg.getLedgerId(), msg.getEntryId());
        LongPair batchSize = pendingAcks.get(position.getLedgerId(), position.getEntryId());
        if (batchSize != null) {
            pendingAcks.remove(position.getLedgerId(), position.getEntryId());
            totalRedeliveryMessages += batchSize.first;
            pendingPositions.add(position);
        }
    }
    addAndGetUnAckedMsgs(this, -totalRedeliveryMessages);
    blockedConsumerOnUnackedMsgs = false;
    if (log.isDebugEnabled()) {
        log.debug("[{}-{}] consumer {} received {} msg-redelivery {}", topicName, subscription, consumerId,
                  totalRedeliveryMessages, pendingPositions.size());
    }
    subscription.redeliverUnacknowledgedMessages(this, pendingPositions);
    msgRedeliver.recordMultipleEvents(totalRedeliveryMessages, totalRedeliveryMessages);
    int numberOfBlockedPermits = PERMITS_RECEIVED_WHILE_CONSUMER_BLOCKED_UPDATER.getAndSet(this, 0);
    // if permitsReceivedWhileConsumerBlocked has been accumulated then pass it to Dispatcher to flow messages
    if (numberOfBlockedPermits > 0) {
        MESSAGE_PERMITS_UPDATER.getAndAdd(this, numberOfBlockedPermits);
        if (log.isDebugEnabled()) {
            log.debug("[{}-{}] Added {} blockedPermits to broker.service.Consumer's messagePermits for consumer {}",
                      topicName, subscription, numberOfBlockedPermits, consumerId);
        }
        subscription.consumerFlow(this, numberOfBlockedPermits);
    }
}

@Override
    public synchronized void redeliverUnacknowledgedMessages(Consumer consumer, List<PositionImpl> positions) {
        positions.forEach(position -> {
            if (addMessageToReplay(position.getLedgerId(), position.getEntryId())) {
                redeliveryTracker.addIfAbsent(position);
            }
        });
        if (log.isDebugEnabled()) {
            log.debug("[{}-{}] Redelivering unacknowledged messages for consumer {}", name, consumer, positions);
        }
        readMoreEntries();
    }

package main;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * @function
 * @date 2021/8/2 17:18
 */
public class FunctionTest {
    private static final Logger log = LoggerFactory.getLogger(FunctionTest.class);

    public static void main(String[] args) throws Exception {
        FunctionTest main = new FunctionTest();
        System.out.println("\n------------------------------------   函数式编程方式  ------------------------------------");
        main.run();
        System.out.println("\n------------------------------------ 经典的异步实现方式 ------------------------------------");
        main.runOld();
    }

    long value = 101;

    void run() throws Exception {
        // 定义任务内容
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            value += 101;
            log.info("supplyAsync, value=" + value);
            return String.format("%d", value);
        });

        Thread.sleep(1000);
        log.info("sleep done");

        // 执行完任务后执行匿名函数
        CompletableFuture<Void> next = future.thenAccept(v -> {
            log.info("thenAccept, value=" + v);
        });

        // 获取任务的执行结果
        String result = future.get();
        log.info("thenAccept, get=" + result);
    }

    void runOld() throws Exception {
        value = 101;
        LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<>();
        // 启动线程执行任务
        task(queue);

        // 阻塞方式等待任务执行结束
        String result = queue.take();
        log.info("thenAccept, get=" + result);
    }

    void task(LinkedBlockingQueue<String> queue) {
        new Thread(() -> {
            String result = "failed";
            try {
                value += 101;
                log.info("supplyAsync, value=" + value);
                result = String.format("%d", value);
            } catch (Exception e) {
                log.error("", e);
            }

            try {
                queue.put(result);
            } catch (InterruptedException e) {
                log.error("", e);
            }
        }).start();
    }
}

@Override
public void initialize(ServiceConfiguration config) throws IOException {
    // 初始化时使用，在broker启动的时候初始化。
}
@Override
public String getAuthMethodName() {
    // 范围认证策略的名称，需要和客户端发送的名称保持一致。
}
@Override
public String authenticate(AuthenticationDataSource authData) throws AuthenticationException {
    // 进行客户端认证，返回客户端的角色名称。
}
@Override
public void close() throws IOException {
    // broker关闭时调用此方法。
}

@Override
public String getAuthMethodName() {
    // 认证策略的名称。
}
@Override
public AuthenticationDataProvider getAuthData() throws PulsarClientException {
   // 返回认证数据持有对象，即另外一个核心接口的实现类。
}
@Override
public void configure(Map<String, String> authParams) {
    // 程序启动时调用，可以得到配置文件中的配置项。
}
@Override
public void start() throws PulsarClientException {
    // 程序启动时调用
}
@Override
public void close() throws IOException {
    // 程序停止时调用
}

@Override
public boolean hasDataForHttp() {
    // http区是否有数据
}
@Override
public Set<Map.Entry<String, String>> getHttpHeaders() throws Exception {
   // 获取http头信息，如果上述方法返回true，则这里要设置http header内容。然后broker端才能根据http header name获取到数据。
}
@Override
public boolean hasDataFromCommand() {
    // 是否有命令数据
}
@Override
public String getCommandData() {
    // 如果上述方法返回true，则这里要返回broker端需要的数据。
}

package auth.server;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
import org.apache.pulsar.broker.authentication.AuthenticationProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.naming.AuthenticationException;
import java.io.IOException;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
/**
 * @function
 * @date 2021/7/27 14:19
 */
public class VVAuthenticationProvider implements AuthenticationProvider {
    private static final Logger log = LoggerFactory.getLogger(VVAuthenticationProvider.class);
    private static final String methodName = "vv_auth_v2";
    private AtomicLong seq = new AtomicLong();
    private String header = "vv_auth";
    @Override
    public void initialize(ServiceConfiguration config) throws IOException {
        log.info(methodName + " initialize" + ", seq=" + seq.incrementAndGet());
        if (config == null) {
            return;
        }
        Set<String> superRoles = config.getSuperUserRoles();
        if (superRoles == null) {
            return;
        }
        for (String role : superRoles) {
            log.info(methodName + " initialize " + role + ", seq=" + seq.incrementAndGet());
        }
    }
    @Override
    public String getAuthMethodName() {
        log.info(methodName + " getAuthMethodName");
        return methodName;
    }
    @Override
    public String authenticate(AuthenticationDataSource authData) throws AuthenticationException {
        log.info(methodName + " authenticate" + ", seq=" + seq.incrementAndGet());
        String roleToken = "unknown";
        if (authData.hasDataFromCommand()) {
            roleToken = authData.getCommandData();
        } else if (authData.hasDataFromHttp()) {
            roleToken = authData.getHttpHeader(header);
        } else {
            throw new AuthenticationException("Authentication data source does not have a role token");
        }
        log.info(methodName + " authenticate " + roleToken + ", seq=" + seq.incrementAndGet());
        return roleToken;
    }
    @Override
    public void close() throws IOException {
        log.info(methodName + " close" + ", seq=" + seq.incrementAndGet());
    }
}

 package auth.client;
import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.client.api.AuthenticationDataProvider;
import org.apache.pulsar.client.api.PulsarClientException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.Map;
/**
 * @function
 * @date 2021/7/27 14:00
 */
public class VVAuthentication implements Authentication {
    private static final Logger log = LoggerFactory.getLogger(VVAuthentication.class);
    private static final String methodName = "vv_auth_v2";
    @Override
    public String getAuthMethodName() {
        log.info(methodName + " getAuthMethodName");
        return methodName;
    }
    @Override
    public AuthenticationDataProvider getAuthData() throws PulsarClientException {
        log.info(methodName + " getAuthData");
        return new VVAuthenticationDataProvider();
    }
    @Override
    public void configure(Map<String, String> authParams) {
        log.info(methodName + " configure");
        if (authParams == null) {
            return;
        }
        authParams.forEach((key, value) -> {
            log.info(methodName + " configure " + key + "=" + value);
        });
    }
    @Override
    public void start() throws PulsarClientException {
        log.info(methodName + " start");
    }
    @Override
    public void close() throws IOException {
        log.info(methodName + " close");
    }
}

package auth.client;
import org.apache.pulsar.client.api.AuthenticationDataProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
/**
 * @function
 * @date 2021/7/27 14:08
 */
public class VVAuthenticationDataProvider implements AuthenticationDataProvider {
    private static final Logger log = LoggerFactory.getLogger(VVAuthenticationDataProvider.class);
    private static final String methodName = "vv_auth_v2";
    private String header = "vv_auth";
    private String token = "vv-role";
    @Override
    public boolean hasDataForHttp() {
        log.info(methodName + " hasDataForHttp");
        return true;
    }
    @Override
    public Set<Map.Entry<String, String>> getHttpHeaders() throws Exception {
        log.info(methodName + " getHttpHeaders");
        Map<String, String> headers = new HashMap<>();
        headers.put(header, token);
        return headers.entrySet();
    }
    @Override
    public boolean hasDataFromCommand() {
        log.info(methodName + " hasDataFromCommand");
        return true;
    }
    @Override
    public String getCommandData() {
        log.info(methodName + " getCommandData");
        return token;
    }
}

package auth;
import auth.client.VVAuthentication;
import org.apache.pulsar.client.api.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
 * @function
 * @date 2021/7/19 10:47
 */
public class AuthTest {
    private static final Logger log = LoggerFactory.getLogger(AuthTest.class);
    public static void main(String[] args) throws Exception {
        AuthTest main = new AuthTest();
        main.run();
    }
    String pulsarUrl = "pulsar://x.x.x.x:6650";
    String topic = "persistent://tenant_vv/ns1/auth_test";
    Authentication authentication = new VVAuthentication();
    void run() throws Exception {
        PulsarClient client = PulsarClient.builder()
                .authentication(authentication)
                .serviceUrl(pulsarUrl)
                .build();
//        send(client);
//        testReader(client);
        consume(client);
        System.out.println("connect successed ");
        client.close();
    }
    void consume(PulsarClient client) throws Exception {
        Consumer consumer = client.newConsumer()
                .topic(topic)
                .subscriptionName("consumer-test")
                .subscribe();
        while (true) {
            Message m = consumer.receive();
            if (m != null) {
                log.info("recv " + new String(m.getData()));
                consumer.acknowledge(m);
            } else {
                break;
            }
        }
    }
    void send(PulsarClient client) throws Exception {
        Producer p = client.newProducer()
                .topic(topic)
                .create();
        for (int i=0; i<10; i++) {
            p.newMessage().key("aaa").value(("hello " + i).getBytes()).send();
            log.info("send " + i);
            Thread.sleep(1000);
        }
        p.flush();
        p.close();
        System.out.println("send done");
    }
    void testReader(PulsarClient client) throws Exception {
        Reader reader = client.newReader()
                .subscriptionName("reader-test")
                .topic(topic)
                .startMessageId(MessageId.earliest)
//                .startMessageId(DefaultImplementation.newMessageId(121493, -1, -1))
                .create();
        while (reader.hasMessageAvailable()) {
            Message msg = reader.readNext();
            log.info("reader recv msg, id=" + msg.getMessageId() + " key=" + msg.getKey() + ", value=" + new String(msg.getData()));
        }
        reader.close();
    }
}

case PARTITIONED_METADATA
case PARTITIONED_METADATA_RESPONSE
case LOOKUP
case LOOKUP_RESPONSE
case ACK
case ACK_RESPONSE
case CLOSE_CONSUMER
case CLOSE_PRODUCER
case CONNECT
case CONNECTED
case ERROR
case FLOW
case MESSAGE
case PRODUCER
case SEND
case SEND_ERROR
case SEND_RECEIPT
case SUBSCRIBE
case SUCCESS
case PRODUCER_SUCCESS
case UNSUBSCRIBE
case SEEK
case PING
case PONG
case REDELIVER_UNACKNOWLEDGED_MESSAGES
case CONSUMER_STATS
case CONSUMER_STATS_RESPONSE
case REACHED_END_OF_TOPIC
case GET_LAST_MESSAGE_ID
case GET_LAST_MESSAGE_ID_RESPONSE
case ACTIVE_CONSUMER_CHANGE
case GET_TOPICS_OF_NAMESPACE
case GET_TOPICS_OF_NAMESPACE_RESPONSE
case GET_SCHEMA
case GET_SCHEMA_RESPONSE
case GET_OR_CREATE_SCHEMA
case GET_OR_CREATE_SCHEMA_RESPONSE
case AUTH_CHALLENGE
case AUTH_RESPONSE
case NEW_TXN
case NEW_TXN_RESPONSE
case ADD_PARTITION_TO_TXN
case ADD_PARTITION_TO_TXN_RESPONSE
case ADD_SUBSCRIPTION_TO_TXN
case ADD_SUBSCRIPTION_TO_TXN_RESPONSE
case END_TXN
case END_TXN_RESPONSE
case END_TXN_ON_PARTITION
case END_TXN_ON_PARTITION_RESPONSE
case END_TXN_ON_SUBSCRIPTION
case END_TXN_ON_SUBSCRIPTION_RESPONSE