/*
 * Copyright 2016 Ignacio del Valle Alles idelvall@brutusin.org.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.brutusin.rpc.websocket;

import java.lang.reflect.Type;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import org.brutusin.json.spi.JsonCodec;
import org.brutusin.json.spi.JsonSchema;
import org.brutusin.rpc.RpcComponent;
import org.springframework.core.ResolvableType;

/**
 *
 * @author Ignacio del Valle Alles idelvall@brutusin.org
 * @param <F> Filter class
 * @param <M> Message class
 */
public abstract class Topic<F, M> extends RpcComponent {

    private final Set<WritableSession> sessions = Collections.synchronizedSet(new HashSet<WritableSession>());

    public final Set<WritableSession> getSubscribers() {
        return sessions;
    }

    public abstract Set<WritableSession> getSubscribers(F filter);

    public final void fire(F filter, M message){
        if (message == null) {
            return;
        }
        JsonSchema messageSchema = JsonCodec.getInstance().getSchema(getMessageType());
        try {
            messageSchema.validate(JsonCodec.getInstance().parse(JsonCodec.getInstance().transform(message)));
        } catch (Exception ex) {
            throw new RuntimeException(ex);
        }
        MessageResponse mr = new MessageResponse();
        mr.setTopic(getId());
        mr.setMessage(message);
        Set<WritableSession> subscribers = getSubscribers(filter);
        synchronized (subscribers) {
            if (subscribers != null) {
                for (WritableSession session : subscribers) {
                    session.sendToPeer(mr);
                }
            }
        }
    }

    public final void subscribe() throws InvalidSubscriptionException {
        WritableSession session = (WritableSession) WebsocketActionContext.getInstance().getSession();
        if (sessions.contains(session)) {
            throw new InvalidSubscriptionException("Current session is already subscribed to this topic");
        }
        sessions.add(session);
        onSubscribe(session);
    }

    public final void unsubscribe() throws InvalidSubscriptionException {
        WritableSession session = (WritableSession) WebsocketActionContext.getInstance().getSession();
        if (!sessions.contains(session)) {
            throw new InvalidSubscriptionException("Current session is not subscribed to this topic");
        }
        sessions.remove(session);
        onUnsubscribe(session);
    }

    protected void onSubscribe(WritableSession session) {
    }

    protected void onUnsubscribe(WritableSession session) {
    }

    public Type geFilterType() {
        return getType(ResolvableType.forClass(Topic.class, getClass()).getGenerics()[0]);
    }

    public Type getMessageType() {
        return getType(ResolvableType.forClass(Topic.class, getClass()).getGenerics()[1]);
    }

}

@@ -1275,95 +1275,344 @@
     
-public final Set%3CWritableSession%3E getSubscribers() %7B%0D%0A        return sessions;%0D%0A    %7D%0D%0A
+/**%0D%0A     * Returns the whole collection of active topic subscribers%0D%0A     *%0D%0A     * @return%0D%0A     */%0D%0A    public final Set%3CWritableSession%3E getSubscribers() %7B%0D%0A        return sessions;%0D%0A    %7D%0D%0A%0D%0A    /**%0D%0A     * Returns the subset of subscribers that satisfy the filtering criteria%0D%0A     *%0D%0A     * @param filter%0D%0A     * @return%0D%0A     */
 %0D%0A  
@@ -1726,16 +1726,17 @@
 message)
+ 
 %7B%0D%0A     
@@ -3359,70 +3359,620 @@
     
-protected void onSubscribe(WritableSession session) %7B%0D%0A    %7D%0D%0A
+/**%0D%0A     * These two methods are auxiliary methods that subclasses can implement to%0D%0A     * be keep track by themselves of the actual subscribers of the topic, and%0D%0A     * create data structures better suited for a efficient topic filtering than%0D%0A     * iterating over the whole collection of subscribers.%0D%0A     *%0D%0A     * See %7B@link #onUnsubscribe(org.brutusin.rpc.websocket.WritableSession)%7D%0D%0A     * @param session%0D%0A     */%0D%0A    protected void onSubscribe(WritableSession session) %7B%0D%0A    %7D%0D%0A%0D%0A    /**%0D%0A     * See %7B@link #onSubscribe(org.brutusin.rpc.websocket.WritableSession)%7D%0D%0A     * @param session %0D%0A     */
 %0D%0A  

improve javadoc