import com.google.firebase.firestore.DocumentChange; //导入依赖的package包/类
@Override
public void onEvent(QuerySnapshot documentSnapshots, FirebaseFirestoreException e) {
    for (DocumentChange change : documentSnapshots.getDocumentChanges()) {
        switch (change.getType()) {
            case ADDED: {
                String groupId = change.getDocument().getId();
                FirebaseFirestore.getInstance()
                        .collection("groups")
                        .document(groupId)
                        .collection("items")
                        .addSnapshotListener(new GroupItemsChangeListener(groupId));
            }
            break;
        }
    }
}
 

import com.google.firebase.firestore.DocumentChange; //导入依赖的package包/类
private void handleListenErrors() {
    // [START handle_listen_errors]
    db.collection("cities")
            .addSnapshotListener(new EventListener<QuerySnapshot>() {
                @Override
                public void onEvent(@Nullable QuerySnapshot snapshots,
                                    @Nullable FirebaseFirestoreException e) {
                    if (e != null) {
                        Log.w(TAG, "listen:error", e);
                        return;
                    }

                    for (DocumentChange dc : snapshots.getDocumentChanges()) {
                        if (dc.getType() == Type.ADDED) {
                            Log.d(TAG, "New city: " + dc.getDocument().getData());
                        }
                    }

                }
            });
    // [END handle_listen_errors]
}
 

import com.google.firebase.firestore.DocumentChange; //导入依赖的package包/类
private void offlineListen(FirebaseFirestore db) {
    // [START offline_listen]
    db.collection("cities").whereEqualTo("state", "CA")
            .addSnapshotListener(new EventListener<QuerySnapshot>() {
                @Override
                public void onEvent(@Nullable QuerySnapshot querySnapshot,
                                    @Nullable FirebaseFirestoreException e) {
                    if (e != null) {
                        Log.w(TAG, "Listen error", e);
                        return;
                    }

                    for (DocumentChange change : querySnapshot.getDocumentChanges()) {
                        if (change.getType() == Type.ADDED) {
                            Log.d(TAG, "New city:" + change.getDocument().getData());
                        }

                        String source = querySnapshot.getMetadata().isFromCache() ?
                                "local cache" : "server";
                        Log.d(TAG, "Data fetched from " + source);
                    }

                }
            });
    // [END offline_listen]
}
 

import com.google.firebase.firestore.DocumentChange; //导入依赖的package包/类
/**
 * Subscribes to the entity states of the given type.
 *
 * <p>The method returns a {@link LiveData} of map (string ID -> entity state). The ID is
 * the {@linkplain io.spine.Identifier#toString(Object) string representation} of
 * the corresponding entity ID.
 *
 * <p>Currently, the element removal is not supported. If a {@link DocumentChange} of type other
 * than {@link DocumentChange.Type#ADDED ADDED} or {@link DocumentChange.Type#MODIFIED MODIFIED}
 * is encountered, an {@link UnsupportedOperationException} is thrown.
 *
 * @param targetType the class of the entity to subscribe to
 * @param <T>        the type of the entity to subscribe to
 * @return an instance of {@link LiveData} for the observers to subscribe to
 */
public <T extends Message> LiveData<Map<String, T>> subscribeTo(Class<T> targetType) {
    checkNotNull(targetType);
    final CollectionReference targetCollection = collectionFor(targetType);
    final MutableLiveData<Map<String, T>> result = new MutableLiveData<>();
    targetCollection.addSnapshotListener((documentSnapshots, error) -> {
        if (error != null) {
            final String errorMsg = format(
                    "Error encountered while listening for the %s state updates.",
                    targetType
            );
            Log.e(TAG, errorMsg, error);
        } else {
            final Parser<T> parser = getParserFor(targetType);
            for (DocumentChange change : documentSnapshots.getDocumentChanges()) {
                deliverUpdate(change, result, parser);
            }
        }
    });
    return result;
}
 

import com.google.firebase.firestore.DocumentChange; //导入依赖的package包/类
/**
 * Delivers the entity state update represented by the given {@link DocumentChange} to
 * the observers of the given {@link LiveData}.
 *
 * @param change      the Firestore document change
 * @param destination the {@link LiveData} publishing the update
 * @param parser      the {@link Parser} for the target entity state type
 * @param <T>         the entity state type
 */
private static <T extends Message>
void deliverUpdate(DocumentChange change,
                   MutableLiveData<Map<String, T>> destination,
                   Parser<T> parser) {
    final DocumentChange.Type type = change.getType();
    final Map<String, T> currentData = destination.getValue();
    final Map<String, T> newData = currentData == null
                                   ? newHashMap()
                                   : newHashMap(currentData);
    final DocumentSnapshot doc = change.getDocument();
    final String id = parseMessageId(doc);
    final T newMessage = parseMessage(doc, parser);

    if (type == ADDED || type == MODIFIED) {
        newData.put(id, newMessage);
    } else {
        throw newIllegalArgumentException("Unexpected document change: %s", type.toString());
    }
    destination.postValue(newData);
}
 

import com.google.firebase.firestore.DocumentChange; //导入依赖的package包/类
@Override
public void onEvent(QuerySnapshot documentSnapshots, FirebaseFirestoreException e) {
    if (e != null) {
        Log.w(TAG, "onEvent:error", e);
        onError(e);
        return;
    }

    // Dispatch the event
    Log.d(TAG, "onEvent:numChanges:" + documentSnapshots.getDocumentChanges().size());
    for (DocumentChange change : documentSnapshots.getDocumentChanges()) {
        switch (change.getType()) {
            case ADDED:
                onDocumentAdded(change);
                break;
            case MODIFIED:
                onDocumentModified(change);
                break;
            case REMOVED:
                onDocumentRemoved(change);
                break;
        }
    }

    onDataChanged();
}
 

import com.google.firebase.firestore.DocumentChange; //导入依赖的package包/类
@Override
public void onEvent(QuerySnapshot snapshots, FirebaseFirestoreException e) {
    if (e != null) {
        notifyOnError(e);
        return;
    }

    // Break down each document event
    List<DocumentChange> changes = snapshots.getDocumentChanges();
    for (DocumentChange change : changes) {
        switch (change.getType()) {
            case ADDED:
                onDocumentAdded(change);
                break;
            case REMOVED:
                onDocumentRemoved(change);
                break;
            case MODIFIED:
                onDocumentModified(change);
                break;
        }
    }

    notifyOnDataChanged();
}
 

import com.google.firebase.firestore.DocumentChange; //导入依赖的package包/类
private void onDocumentModified(DocumentChange change) {
    if (change.getOldIndex() == change.getNewIndex()) {
        // Document modified only
        mSnapshots.set(change.getNewIndex(), change.getDocument());
        notifyOnChildChanged(ChangeEventType.CHANGED, change.getDocument(),
                change.getNewIndex(), change.getNewIndex());
    } else {
        // Document moved and possibly also modified
        mSnapshots.remove(change.getOldIndex());
        mSnapshots.add(change.getNewIndex(), change.getDocument());

        notifyOnChildChanged(ChangeEventType.MOVED, change.getDocument(),
                change.getNewIndex(), change.getOldIndex());
        notifyOnChildChanged(ChangeEventType.CHANGED, change.getDocument(),
                change.getNewIndex(), change.getNewIndex());
    }
}
 

import com.google.firebase.firestore.DocumentChange; //导入依赖的package包/类
@Override
public void onEvent(QuerySnapshot documentSnapshots, FirebaseFirestoreException e) {
    for (DocumentChange change : documentSnapshots.getDocumentChanges()) {
        switch (change.getType()) {
            case ADDED: {
                Group group = new Group();
                group.setId(change.getDocument().getId());
                userGroups.add(group);
                adapter.setUserGroups(userGroups);

                FirebaseFirestore.getInstance()
                        .collection("groups")
                        .whereEqualTo("parentGroup", change.getDocument().getId())
                        .addSnapshotListener(CoursesFragment.this);
            }
            break;
            case REMOVED: {
                for (int i = 0; i < userGroups.size(); i++)
                    if (userGroups.get(i).getId().equals(change.getDocument().getId())) {
                        userGroups.remove(i);
                        adapter.notifyDataSetChanged();
                        return;
                    }
            }
            break;
        }
    }
}
 

import com.google.firebase.firestore.DocumentChange; //导入依赖的package包/类
@Override
public void onEvent(QuerySnapshot documentSnapshots,
                    FirebaseFirestoreException e) {

    // Handle errors
    if (e != null) {
        Log.w(TAG, "onEvent:error", e);
        return;
    }

    // Dispatch the event
    for (DocumentChange change : documentSnapshots.getDocumentChanges()) {
        // Snapshot of the changed document
        DocumentSnapshot snapshot = change.getDocument();

        switch (change.getType()) {
            case ADDED:
                onDocumentAdded(change);
                break;
            case MODIFIED:
                onDocumentModified(change);
                break;
            case REMOVED:
                onDocumentRemoved(change);
                break;
        }
    }

    onDataChanged();
}
 

import com.google.firebase.firestore.DocumentChange; //导入依赖的package包/类
protected void onDocumentModified(DocumentChange change) {
    if (change.getOldIndex() == change.getNewIndex()) {
        // Item changed but remained in same position
        mSnapshots.set(change.getOldIndex(), change.getDocument());
        notifyItemChanged(change.getOldIndex());
    } else {
        // Item changed and changed position
        mSnapshots.remove(change.getOldIndex());
        mSnapshots.add(change.getNewIndex(), change.getDocument());
        notifyItemMoved(change.getOldIndex(), change.getNewIndex());
    }
}
 

import com.google.firebase.firestore.DocumentChange; //导入依赖的package包/类
@Override
public void subscribe(final ObservableEmitter<DocumentChange> emitter) throws Exception {
    final EventListener<QuerySnapshot> listener = new EventListener<QuerySnapshot>() {
        @Override
        public void onEvent(QuerySnapshot querySnapshot, FirebaseFirestoreException e) {

            if (!emitter.isDisposed()) {
                if (e == null) {
                    for (DocumentChange change : querySnapshot.getDocumentChanges()) {
                        emitter.onNext(change);
                    }

                } else {
                    emitter.onError(e);
                }
            }

        }
    };

    registration = query.addSnapshotListener(listener);

    emitter.setDisposable(Disposables.fromAction(new Action() {
        @Override
        public void run() throws Exception {
            registration.remove();
        }
    }));
}
 

import com.google.firebase.firestore.DocumentChange; //导入依赖的package包/类
private void listenToDiffs() {
    // [START listen_diffs]
    db.collection("cities")
            .whereEqualTo("state", "CA")
            .addSnapshotListener(new EventListener<QuerySnapshot>() {
                @Override
                public void onEvent(@Nullable QuerySnapshot snapshots,
                                    @Nullable FirebaseFirestoreException e) {
                    if (e != null) {
                        Log.w(TAG, "listen:error", e);
                        return;
                    }

                    for (DocumentChange dc : snapshots.getDocumentChanges()) {
                        switch (dc.getType()) {
                            case ADDED:
                                Log.d(TAG, "New city: " + dc.getDocument().getData());
                                break;
                            case MODIFIED:
                                Log.d(TAG, "Modified city: " + dc.getDocument().getData());
                                break;
                            case REMOVED:
                                Log.d(TAG, "Removed city: " + dc.getDocument().getData());
                                break;
                        }
                    }

                }
            });
    // [END listen_diffs]
}
 

import com.google.firebase.firestore.DocumentChange; //导入依赖的package包/类
private void listenState() {
    // [START listen_state]
    db.collection("cities")
            .whereEqualTo("state", "CA")
            .addSnapshotListener(new EventListener<QuerySnapshot>() {
                @Override
                public void onEvent(@Nullable QuerySnapshot snapshots,
                                    @Nullable FirebaseFirestoreException e) {

                    if (e != null) {
                        Log.w(TAG, "listen:error", e);
                        return;
                    }

                    for (DocumentChange dc : snapshots.getDocumentChanges()) {
                        if (dc.getType() == Type.ADDED) {
                            Log.d(TAG, "New city: " + dc.getDocument().getData());
                        }
                    }

                    if (!snapshots.getMetadata().isFromCache()) {
                        Log.d(TAG, "Got initial state.");
                    }
                }
            });
    // [END listen_state]
}
 

import com.google.firebase.firestore.DocumentChange; //导入依赖的package包/类
protected void onDocumentAdded(DocumentChange change) {
    mSnapshots.add(change.getNewIndex(), change.getDocument());
    notifyItemInserted(change.getNewIndex());
}
 

import com.google.firebase.firestore.DocumentChange; //导入依赖的package包/类
protected void onDocumentRemoved(DocumentChange change) {
    mSnapshots.remove(change.getOldIndex());
    notifyItemRemoved(change.getOldIndex());
}
 

import com.google.firebase.firestore.DocumentChange; //导入依赖的package包/类
@NonNull
@CheckResult
public static Observable<DocumentChange> documentChanges(@NonNull Query query) {
    return Observable.create(new DocumentChangesOnSubscribe(query));
}
 

import com.google.firebase.firestore.DocumentChange; //导入依赖的package包/类
private void onDocumentAdded(DocumentChange change) {
    mSnapshots.add(change.getNewIndex(), change.getDocument());
    notifyOnChildChanged(ChangeEventType.ADDED, change.getDocument(), change.getNewIndex(), -1);
}
 

import com.google.firebase.firestore.DocumentChange; //导入依赖的package包/类
private void onDocumentRemoved(DocumentChange change) {
    mSnapshots.remove(change.getOldIndex());
    notifyOnChildChanged(
            ChangeEventType.REMOVED, change.getDocument(), -1, change.getOldIndex());
}