@Override @NotNull @SuppressWarnings({"unchecked"}) public <L> L syncPublisher(@NotNull final Topic<L> topic) { checkNotDisposed(); L publisher = (L)mySyncPublishers.get(topic); if (publisher == null) { final Class<L> listenerClass = topic.getListenerClass(); InvocationHandler handler = new InvocationHandler() { @Override public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { sendMessage(new Message(topic, method, args)); return NA; } }; publisher = (L)Proxy.newProxyInstance(listenerClass.getClassLoader(), new Class[]{listenerClass}, handler); publisher = (L)ConcurrencyUtil.cacheOrGet(mySyncPublishers, topic, publisher); } return publisher; }
@Override @NotNull @SuppressWarnings({"unchecked"}) public <L> L asyncPublisher(@NotNull final Topic<L> topic) { checkNotDisposed(); L publisher = (L)myAsyncPublishers.get(topic); if (publisher == null) { final Class<L> listenerClass = topic.getListenerClass(); InvocationHandler handler = new InvocationHandler() { @Override public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { postMessage(new Message(topic, method, args)); return NA; } }; publisher = (L)Proxy.newProxyInstance(listenerClass.getClassLoader(), new Class[]{listenerClass}, handler); publisher = (L)ConcurrencyUtil.cacheOrGet(myAsyncPublishers, topic, publisher); } return publisher; }
private void calcSubscribers(Topic topic, List<MessageBusConnectionImpl> result) { final List<MessageBusConnectionImpl> topicSubscribers = mySubscribers.get(topic); if (topicSubscribers != null) { result.addAll(topicSubscribers); } Topic.BroadcastDirection direction = topic.getBroadcastDirection(); if (direction == Topic.BroadcastDirection.TO_CHILDREN) { for (MessageBusImpl childBus : myChildBuses) { childBus.calcSubscribers(topic, result); } } if (direction == Topic.BroadcastDirection.TO_PARENT && myParentBus != null) { myParentBus.calcSubscribers(topic, result); } }
private void postMessage(Message message) { checkNotDisposed(); final Topic topic = message.getTopic(); List<MessageBusConnectionImpl> topicSubscribers = mySubscriberCache.get(topic); if (topicSubscribers == null) { topicSubscribers = new SmartList<MessageBusConnectionImpl>(); calcSubscribers(topic, topicSubscribers); mySubscriberCache.put(topic, topicSubscribers); } if (!topicSubscribers.isEmpty()) { for (MessageBusConnectionImpl subscriber : topicSubscribers) { subscriber.getBus().myMessageQueue.get().offer(new DeliveryJob(subscriber, message)); subscriber.getBus().notifyPendingJobChange(1); subscriber.scheduleMessageDelivery(message); } } }
Tool(Project project, String command, ToolKey key, TextFieldWithBrowseButton pathField, RawCommandLineEditor flagsField, JButton autoFindButton, JTextField versionField, String versionParam, @Nullable Topic<SettingsChangeNotifier> topic) { this.project = project; this.command = command; this.key = key; this.pathField = pathField; this.flagsField = flagsField; this.versionField = versionField; this.versionParam = versionParam; this.autoFindButton = autoFindButton; this.topic = topic; this.publisher = topic == null ? null : project.getMessageBus().syncPublisher(topic); this.propertyFields = Arrays.asList( new PropertyField(key.pathKey, pathField), new PropertyField(key.flagsKey, flagsField)); GuiUtil.addFolderListener(pathField, command); GuiUtil.addApplyPathAction(autoFindButton, pathField, command); updateVersion(); }
@Override public void beforeEach(ExtensionContext context) { IdeaMocksImpl ideaMocks = new IdeaMocksImpl(); Project project = mock(Project.class); MessageBus messageBus = mock(MessageBus.class); when(project.getMessageBus()).thenReturn(messageBus); when(messageBus.syncPublisher(any(Topic.class))).thenAnswer(invocation -> { Topic topic = invocation.getArgument(0); Class<?> listenerClass = topic.getListenerClass(); if (ideaMocks.hasMockListener(listenerClass)) { return ideaMocks.getMockListener(listenerClass); } else { return ideaMocks.mockListener(listenerClass); } }); Store store = context.getStore(NS); store.put(Project.class, project); store.put(MessageBus.class, messageBus); store.put(IdeaMocks.class, ideaMocks); }
private void postMessage(Message message) { checkNotDisposed(); final Topic topic = message.getTopic(); final List<MessageBusConnectionImpl> topicSubscribers = mySubscribers.get(topic); if (topicSubscribers != null) { Queue<DeliveryJob> queue = myMessageQueue.get(); for (MessageBusConnectionImpl subscriber : topicSubscribers) { queue.offer(new DeliveryJob(subscriber, message)); subscriber.scheduleMessageDelivery(message); } } Topic.BroadcastDirection direction = topic.getBroadcastDirection(); if (direction == Topic.BroadcastDirection.TO_CHILDREN) { for (MessageBusImpl childBus : myChildBuses) { childBus.postMessage(message); } } if (direction == Topic.BroadcastDirection.TO_PARENT && myParentBus != null) { myParentBus.postMessage(message); } }
@Override @Nonnull @SuppressWarnings("unchecked") public <L> L syncPublisher(@Nonnull final Topic<L> topic) { checkNotDisposed(); L publisher = (L)mySyncPublishers.get(topic); if (publisher == null) { final Class<L> listenerClass = topic.getListenerClass(); InvocationHandler handler = new InvocationHandler() { @Override public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { sendMessage(new Message(topic, method, args)); return NA; } }; publisher = (L)Proxy.newProxyInstance(listenerClass.getClassLoader(), new Class[]{listenerClass}, handler); publisher = (L)ConcurrencyUtil.cacheOrGet(mySyncPublishers, topic, publisher); } return publisher; }
@Override @Nonnull @SuppressWarnings("unchecked") public <L> L asyncPublisher(@Nonnull final Topic<L> topic) { checkNotDisposed(); L publisher = (L)myAsyncPublishers.get(topic); if (publisher == null) { final Class<L> listenerClass = topic.getListenerClass(); InvocationHandler handler = new InvocationHandler() { @Override public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { postMessage(new Message(topic, method, args)); return NA; } }; publisher = (L)Proxy.newProxyInstance(listenerClass.getClassLoader(), new Class[]{listenerClass}, handler); publisher = (L)ConcurrencyUtil.cacheOrGet(myAsyncPublishers, topic, publisher); } return publisher; }
public static <T> void runOnSyncPublisher(final Project project, final Topic<T> topic, final Consumer<T> listener) { final Application application = ApplicationManager.getApplication(); final Runnable runnable = createPublisherRunnable(project, topic, listener); if (application.isDispatchThread()) { runnable.run(); } else { application.runReadAction(runnable); } }
private static <T> Runnable createPublisherRunnable(final Project project, final Topic<T> topic, final Consumer<T> listener) { return new Runnable() { @Override public void run() { if (project.isDisposed()) throw new ProcessCanceledException(); listener.consume(project.getMessageBus().syncPublisher(topic)); } }; }
public static <T> void invokeLaterIfNeededOnSyncPublisher(final Project project, final Topic<T> topic, final Consumer<T> listener) { final Application application = ApplicationManager.getApplication(); final Runnable runnable = createPublisherRunnable(project, topic, listener); if (application.isDispatchThread()) { runnable.run(); } else { application.invokeLater(runnable); } }
@Override public <L> void subscribe(@NotNull Topic<L> topic, @NotNull L handler) throws IllegalStateException { synchronized (myPendingMessages) { if (mySubscriptions.get(topic) != null) { throw new IllegalStateException("Subscription to " + topic + " already exists"); } mySubscriptions = mySubscriptions.plus(topic, handler); } myBus.notifyOnSubscription(this, topic); }
@Override @SuppressWarnings("unchecked") public <L> void subscribe(@NotNull Topic<L> topic) throws IllegalStateException { if (myDefaultHandler == null) { throw new IllegalStateException("Connection must have default handler installed prior to any anonymous subscriptions. " + "Target topic: " + topic); } if (topic.getListenerClass().isInstance(myDefaultHandler)) { throw new IllegalStateException("Can't subscribe to the topic '" + topic +"'. Default handler has incompatible type - expected: '" + topic.getListenerClass() + "', actual: '" + myDefaultHandler.getClass() + "'"); } subscribe(topic, (L)myDefaultHandler); }
void notifyOnSubscription(final MessageBusConnectionImpl connection, final Topic topic) { checkNotDisposed(); List<MessageBusConnectionImpl> topicSubscribers = mySubscribers.get(topic); if (topicSubscribers == null) { topicSubscribers = ContainerUtil.createLockFreeCopyOnWriteList(); topicSubscribers = ConcurrencyUtil.cacheOrGet(mySubscribers, topic, topicSubscribers); } topicSubscribers.add(connection); getRootBus().clearSubscriberCache(); }
void notifyOnSubscription(final MessageBusConnectionImpl connection, final Topic topic) { checkNotDisposed(); List<MessageBusConnectionImpl> topicSubscribers = mySubscribers.get(topic); if (topicSubscribers == null) { topicSubscribers = ContainerUtil.createLockFreeCopyOnWriteList(); topicSubscribers = ConcurrencyUtil.cacheOrGet(mySubscribers, topic, topicSubscribers); } topicSubscribers.add(connection); }
/** * Wraps {@link MessageBus#syncPublisher(Topic)} in a dispose check, * and throws a {@link ProcessCanceledException} if the project is disposed, * instead of throwing an assertion which would happen otherwise. * * @see #syncPublisher(Topic) */ @Nonnull public static <L> L syncPublisher(@Nonnull Project project, @Nonnull Topic<L> topic) throws ProcessCanceledException { return ReadAction.compute(() -> { if (project.isDisposed()) throw new ProcessCanceledException(); return project.getMessageBus().syncPublisher(topic); }); }
/** * Wraps {@link MessageBus#syncPublisher(Topic)} in a dispose check, * and throws a {@link ProcessCanceledException} if the application is disposed, * instead of throwing an assertion which would happen otherwise. * * @see #syncPublisher(Project, Topic) */ @Nonnull public static <L> L syncPublisher(@Nonnull Topic<L> topic) throws ProcessCanceledException { return ReadAction.compute(() -> { if (ApplicationManager.getApplication().isDisposed()) throw new ProcessCanceledException(); return ApplicationManager.getApplication().getMessageBus().syncPublisher(topic); }); }
@Override public <L> void subscribe(@Nonnull Topic<L> topic, @Nonnull L handler) throws IllegalStateException { synchronized (myPendingMessages) { if (mySubscriptions.get(topic) != null) { throw new IllegalStateException("Subscription to " + topic + " already exists"); } mySubscriptions = mySubscriptions.plus(topic, handler); } myBus.notifyOnSubscription(this, topic); }
@Override public <L> void subscribe(@Nonnull Topic<L> topic) throws IllegalStateException { if (myDefaultHandler == null) { throw new IllegalStateException("Connection must have default handler installed prior to any anonymous subscriptions. " + "Target topic: " + topic); } if (topic.getListenerClass().isInstance(myDefaultHandler)) { throw new IllegalStateException("Can't subscribe to the topic '" + topic +"'. Default handler has incompatible type - expected: '" + topic.getListenerClass() + "', actual: '" + myDefaultHandler.getClass() + "'"); } //noinspection unchecked subscribe(topic, (L)myDefaultHandler); }
boolean containsMessage(@Nonnull Topic topic) { for (Message message : myPendingMessages.get()) { if (message.getTopic() == topic) { return true; } } return false; }
@Override public boolean hasUndeliveredEvents(@Nonnull Topic<?> topic) { if (!isDispatchingAnything()) return false; for (MessageBusConnectionImpl connection : getTopicSubscribers(topic)) { if (connection.containsMessage(topic)) { return true; } } return false; }
@Nonnull private List<MessageBusConnectionImpl> getTopicSubscribers(Topic topic) { List<MessageBusConnectionImpl> topicSubscribers = mySubscriberCache.get(topic); if (topicSubscribers == null) { topicSubscribers = new SmartList<MessageBusConnectionImpl>(); calcSubscribers(topic, topicSubscribers); mySubscriberCache.put(topic, topicSubscribers); } return topicSubscribers; }
void notifyOnSubscription(@Nonnull MessageBusConnectionImpl connection, @Nonnull Topic<?> topic) { checkNotDisposed(); List<MessageBusConnectionImpl> topicSubscribers = mySubscribers.get(topic); if (topicSubscribers == null) { topicSubscribers = ContainerUtil.createLockFreeCopyOnWriteList(); topicSubscribers = ConcurrencyUtil.cacheOrGet(mySubscribers, topic, topicSubscribers); } topicSubscribers.add(connection); rootBus().clearSubscriberCache(); }
protected AbstractExternalSystemSettings(@NotNull Topic<L> topic, @NotNull Project project) { myChangesTopic = topic; myProject = project; Disposer.register(project, this); }
@NotNull public Topic<L> getChangesTopic() { return myChangesTopic; }
public Message(@NotNull Topic topic, @NotNull Method listenerMethod, Object[] args) { myTopic = topic; listenerMethod.setAccessible(true); myListenerMethod = listenerMethod; myArgs = args; }
@NotNull public Topic getTopic() { return myTopic; }
public MessageListenerList(@NotNull MessageBus messageBus, @NotNull Topic<T> topic) { myTopic = topic; myMessageBus = messageBus; }