macltok

public

Star0

Watch0

20 commits 3 branches 0 tags 0 releases3.0 MiB

#![cfg(feature = "mirror")]

use std::time::Duration;

use futures_util::StreamExt;
use macltok::{
    ClientHandle, Command, Engine, EngineConfig, Event, Mirror, MirrorHandle, StateChange,
    Subscription, TimelineChange, TimelineCommand, TimelineDiff, TimelineId, TimelineItem, Update,
    UploadOutcome,
};
use matrix_sdk::{
    ruma::{OwnedRoomId, TransactionId, api::client::sync::sync_events::v5, event_id, room_id},
    test_utils::mocks::MatrixMockServer,
};
use tokio::sync::mpsc;

struct Harness {
    handle: ClientHandle,
    mirror: MirrorHandle,
    inject: mpsc::UnboundedSender<Event>,
    room_id: OwnedRoomId,
    _server: MatrixMockServer,
}

/// the mirror reads the real engine's stream with synthetic events mixed in, so a test can
/// hand it a diff the engine never would
async fn start(config: EngineConfig, capacity: usize) -> Harness {
    let server = MatrixMockServer::new().await;
    server
        .mock_sliding_sync()
        .ok(v5::Response::new("1".to_owned()))
        .mount()
        .await;
    let client = server.client_builder().build().await;
    let room_id = room_id!("!mirror:localhost");
    server.mock_room_state_encryption().plain().mount().await;
    server.sync_joined_room(&client, room_id).await;
    let (engine, handle, events) =
        Engine::from_client(client, config.with_initial_history_limit(0))
            .await
            .unwrap();
    tokio::spawn(engine.run());
    let (inject, mut injected) = mpsc::unbounded_channel();
    let injected = futures_util::stream::poll_fn(move |context| injected.poll_recv(context));
    let (mirror, mirror_handle) = Mirror::with_capacity(
        handle.clone(),
        futures_util::stream::select(events, injected),
        capacity,
    );
    tokio::spawn(mirror.run());
    Harness {
        handle,
        mirror: mirror_handle,
        inject,
        room_id: room_id.to_owned(),
        _server: server,
    }
}

async fn next_matching(
    subscription: &mut Subscription,
    predicate: impl Fn(&StateChange) -> bool,
) -> Update {
    tokio::time::timeout(Duration::from_secs(5), async {
        loop {
            let update = subscription.next().await.expect("the mirror stopped");
            if predicate(&update.change) {
                return update;
            }
        }
    })
    .await
    .expect("expected update did not arrive")
}

fn timeline_change(
    change: &StateChange,
    id: &TimelineId,
    predicate: impl Fn(&TimelineChange) -> bool,
) -> bool {
    matches!(change, StateChange::Timeline { timeline_id, change } if timeline_id == id && predicate(change))
}

fn item(timestamp_ms: u64) -> TimelineItem {
    TimelineItem::DayDivider { timestamp_ms }
}

fn reset(timeline_id: &TimelineId, items: Vec<TimelineItem>) -> Event {
    Event::TimelineReset {
        timeline_id: timeline_id.clone(),
        items: items.into(),
    }
}

/// the room never opens, so every resync is refused and the divergence holds still to be looked at
#[tokio::test]
async fn diverged_batch_keeps_last_good_items_until_a_reset() {
    let harness = start(EngineConfig::default(), 256).await;
    let timeline_id = TimelineId::from(harness.room_id.clone());
    let mut updates = harness.mirror.subscribe();
    harness
        .inject
        .send(reset(&timeline_id, vec![item(1), item(2)]))
        .unwrap();
    for diffs in [
        vec![
            TimelineDiff::PushBack { value: item(3) },
            TimelineDiff::Remove { index: 7 },
        ],
        vec![TimelineDiff::PushBack { value: item(4) }],
    ] {
        harness
            .inject
            .send(Event::TimelineChanged {
                timeline_id: timeline_id.clone(),
                diffs,
            })
            .unwrap();
    }
    next_matching(&mut updates, |change| {
        timeline_change(change, &timeline_id, |change| {
            matches!(change, TimelineChange::Diverged)
        })
    })
    .await;
    let snapshot = harness.mirror.snapshot();
    let timeline = &snapshot.timelines[&timeline_id];
    assert!(timeline.diverged);
    assert_eq!(timeline.items, [item(1), item(2)].into_iter().collect());

    harness
        .inject
        .send(reset(&timeline_id, vec![item(5)]))
        .unwrap();
    let update = next_matching(&mut updates, |change| {
        timeline_change(change, &timeline_id, |_| true)
    })
    .await;
    assert!(
        timeline_change(&update.change, &timeline_id, |change| {
            matches!(change, TimelineChange::Reset { .. })
        }),
        "a diverged timeline took {update:?}"
    );
    let snapshot = harness.mirror.snapshot();
    let timeline = &snapshot.timelines[&timeline_id];
    assert!(!timeline.diverged);
    assert_eq!(timeline.items, [item(5)].into_iter().collect());
}

#[tokio::test]
async fn room_and_thread_timelines_stay_apart() {
    let harness = start(EngineConfig::default(), 256).await;
    let room = TimelineId::from(harness.room_id.clone());
    let thread = TimelineId::thread(
        harness.room_id.clone(),
        event_id!("$root:localhost").to_owned(),
    );
    let mut updates = harness.mirror.subscribe();
    harness.inject.send(reset(&room, vec![item(1)])).unwrap();
    harness.inject.send(reset(&thread, vec![item(2)])).unwrap();
    harness
        .inject
        .send(Event::TimelineChanged {
            timeline_id: thread.clone(),
            diffs: vec![TimelineDiff::PushBack { value: item(3) }],
        })
        .unwrap();
    next_matching(&mut updates, |change| {
        timeline_change(change, &thread, |change| {
            matches!(change, TimelineChange::Changed { .. })
        })
    })
    .await;
    let snapshot = harness.mirror.snapshot();
    assert_eq!(
        snapshot.timelines[&room].items,
        [item(1)].into_iter().collect()
    );
    assert_eq!(
        snapshot.timelines[&thread].items,
        [item(2), item(3)].into_iter().collect()
    );
}

#[tokio::test]
async fn late_subscriber_gets_state_then_every_update_after_it() {
    let harness = start(EngineConfig::default(), 1024).await;
    let transaction_id = TransactionId::new();
    let progress = |uploaded_bytes| Event::MediaUploadProgress {
        room_id: harness.room_id.clone(),
        transaction_id: transaction_id.clone(),
        uploaded_bytes,
        total_bytes: 100,
        complete: false,
    };
    for uploaded_bytes in 1..=50 {
        harness.inject.send(progress(uploaded_bytes)).unwrap();
    }
    let mut updates = harness.mirror.subscribe();
    for uploaded_bytes in 51..=100 {
        harness.inject.send(progress(uploaded_bytes)).unwrap();
    }

    let first = updates.next().await.unwrap();
    let StateChange::State(state) = &first.change else {
        panic!("a subscription opened with {first:?}");
    };
    let mut seen = state
        .uploads
        .get(&transaction_id)
        .map_or(0, |upload| upload.uploaded_bytes);
    let mut sequence = first.sequence;
    while seen < 100 {
        let update = next_matching(&mut updates, |_| true).await;
        assert_eq!(update.sequence, sequence + 1);
        sequence = update.sequence;
        if let StateChange::Upload { upload, .. } = update.change {
            assert_eq!(upload.uploaded_bytes, seen + 1);
            seen += 1;
        }
    }
}

#[tokio::test]
async fn lagging_subscriber_starts_over_from_a_fresh_state() {
    let harness = start(EngineConfig::default(), 4).await;
    let transaction_id = TransactionId::new();
    let mut updates = harness.mirror.subscribe();
    assert!(matches!(
        updates.next().await.unwrap().change,
        StateChange::State(_)
    ));
    for uploaded_bytes in 1..=20 {
        harness
            .inject
            .send(Event::MediaUploadProgress {
                room_id: harness.room_id.clone(),
                transaction_id: transaction_id.clone(),
                uploaded_bytes,
                total_bytes: 20,
                complete: uploaded_bytes == 20,
            })
            .unwrap();
    }
    tokio::time::timeout(Duration::from_secs(5), async {
        while !harness
            .mirror
            .snapshot()
            .uploads
            .get(&transaction_id)
            .is_some_and(|upload| upload.complete)
        {
            tokio::task::yield_now().await;
        }
    })
    .await
    .expect("the mirror never caught up");

    let update = updates.next().await.unwrap();
    let StateChange::State(state) = &update.change else {
        panic!("a lagged subscription carried on with {update:?}");
    };
    assert!(state.uploads[&transaction_id].complete);
}

#[tokio::test]
async fn resync_does_not_deadlock_on_full_queues() {
    let config = EngineConfig::default()
        .with_command_capacity(1)
        .with_event_capacity(1);
    let harness = start(config, 1024).await;
    let timeline_id = TimelineId::from(harness.room_id.clone());
    let mut updates = harness.mirror.subscribe();
    harness
        .handle
        .open_room(harness.room_id.clone())
        .await
        .unwrap();
    next_matching(&mut updates, |change| {
        timeline_change(change, &timeline_id, |change| {
            matches!(change, TimelineChange::Reset { .. })
        })
    })
    .await;

    let flood = tokio::spawn({
        let handle = harness.handle.clone();
        let timeline_id = timeline_id.clone();
        async move {
            loop {
                let command = Command::Timeline(TimelineCommand::Paginate {
                    timeline_id: timeline_id.clone(),
                    limit: 1,
                });
                if handle.send(command).await.is_err() {
                    return;
                }
            }
        }
    });
    for _ in 0..10 {
        harness
            .inject
            .send(Event::TimelineChanged {
                timeline_id: timeline_id.clone(),
                diffs: vec![TimelineDiff::Remove { index: 1000 }],
            })
            .unwrap();
        next_matching(&mut updates, |change| {
            timeline_change(change, &timeline_id, |change| {
                matches!(change, TimelineChange::Diverged)
            })
        })
        .await;
        next_matching(&mut updates, |change| {
            timeline_change(change, &timeline_id, |change| {
                matches!(change, TimelineChange::Reset { .. })
            })
        })
        .await;
    }
    flood.abort();
}

#[tokio::test]
async fn closing_a_timeline_evicts_it() {
    let harness = start(EngineConfig::default(), 256).await;
    let timeline_id = TimelineId::from(harness.room_id.clone());
    let mut updates = harness.mirror.subscribe();
    harness
        .handle
        .open_room(harness.room_id.clone())
        .await
        .unwrap();
    harness
        .handle
        .close_room(harness.room_id.clone())
        .await
        .unwrap();
    next_matching(&mut updates, |change| {
        timeline_change(change, &timeline_id, |change| {
            matches!(change, TimelineChange::Closed)
        })
    })
    .await;
    assert!(
        !harness
            .mirror
            .snapshot()
            .timelines
            .contains_key(&timeline_id)
    );
}

#[tokio::test]
async fn an_upload_ends_on_finished() {
    let harness = start(EngineConfig::default(), 256).await;
    let transaction_id = TransactionId::new();
    let mut updates = harness.mirror.subscribe();
    harness
        .inject
        .send(Event::MediaUploadProgress {
            room_id: harness.room_id.clone(),
            transaction_id: transaction_id.clone(),
            uploaded_bytes: 5,
            total_bytes: 10,
            complete: false,
        })
        .unwrap();
    next_matching(&mut updates, |change| {
        matches!(change, StateChange::Upload { .. })
    })
    .await;
    assert!(
        harness
            .mirror
            .snapshot()
            .uploads
            .contains_key(&transaction_id)
    );
    harness
        .inject
        .send(Event::MediaUploadFinished {
            room_id: harness.room_id.clone(),
            transaction_id: transaction_id.clone(),
            outcome: UploadOutcome::Cancelled,
        })
        .unwrap();
    next_matching(&mut updates, |change| {
        matches!(change, StateChange::UploadFinished { .. })
    })
    .await;
    assert!(harness.mirror.snapshot().uploads.is_empty());
}

1	#![cfg(feature = "mirror")]
2
3	use std::time::Duration;
4
5	use futures_util::StreamExt;
6	use macltok::{
7	ClientHandle, Command, Engine, EngineConfig, Event, Mirror, MirrorHandle, StateChange,
8	Subscription, TimelineChange, TimelineCommand, TimelineDiff, TimelineId, TimelineItem, Update,
9	UploadOutcome,
10	};
11	use matrix_sdk::{
12	ruma::{OwnedRoomId, TransactionId, api::client::sync::sync_events::v5, event_id, room_id},
13	test_utils::mocks::MatrixMockServer,
14	};
15	use tokio::sync::mpsc;
16
17	struct Harness {
18	handle: ClientHandle,
19	mirror: MirrorHandle,
20	inject: mpsc::UnboundedSender<Event>,
21	room_id: OwnedRoomId,
22	_server: MatrixMockServer,
23	}
24
25	/// the mirror reads the real engine's stream with synthetic events mixed in, so a test can
26	/// hand it a diff the engine never would
27	async fn start(config: EngineConfig, capacity: usize) -> Harness {
28	let server = MatrixMockServer::new().await;
29	server
30	.mock_sliding_sync()
31	.ok(v5::Response::new("1".to_owned()))
32	.mount()
33	.await;
34	let client = server.client_builder().build().await;
35	let room_id = room_id!("!mirror:localhost");
36	server.mock_room_state_encryption().plain().mount().await;
37	server.sync_joined_room(&client, room_id).await;
38	let (engine, handle, events) =
39	Engine::from_client(client, config.with_initial_history_limit(0))
40	.await
41	.unwrap();
42	tokio::spawn(engine.run());
43	let (inject, mut injected) = mpsc::unbounded_channel();
44	let injected = futures_util::stream::poll_fn(move \|context\| injected.poll_recv(context));
45	let (mirror, mirror_handle) = Mirror::with_capacity(
46	handle.clone(),
47	futures_util::stream::select(events, injected),
48	capacity,
49	);
50	tokio::spawn(mirror.run());
51	Harness {
52	handle,
53	mirror: mirror_handle,
54	inject,
55	room_id: room_id.to_owned(),
56	_server: server,
57	}
58	}
59
60	async fn next_matching(
61	subscription: &mut Subscription,
62	predicate: impl Fn(&StateChange) -> bool,
63	) -> Update {
64	tokio::time::timeout(Duration::from_secs(5), async {
65	loop {
66	let update = subscription.next().await.expect("the mirror stopped");
67	if predicate(&update.change) {
68	return update;
69	}
70	}
71	})
72	.await
73	.expect("expected update did not arrive")
74	}
75
76	fn timeline_change(
77	change: &StateChange,
78	id: &TimelineId,
79	predicate: impl Fn(&TimelineChange) -> bool,
80	) -> bool {
81	matches!(change, StateChange::Timeline { timeline_id, change } if timeline_id == id && predicate(change))
82	}
83
84	fn item(timestamp_ms: u64) -> TimelineItem {
85	TimelineItem::DayDivider { timestamp_ms }
86	}
87
88	fn reset(timeline_id: &TimelineId, items: Vec<TimelineItem>) -> Event {
89	Event::TimelineReset {
90	timeline_id: timeline_id.clone(),
91	items: items.into(),
92	}
93	}
94
95	/// the room never opens, so every resync is refused and the divergence holds still to be looked at
96	#[tokio::test]
97	async fn diverged_batch_keeps_last_good_items_until_a_reset() {
98	let harness = start(EngineConfig::default(), 256).await;
99	let timeline_id = TimelineId::from(harness.room_id.clone());
100	let mut updates = harness.mirror.subscribe();
101	harness
102	.inject
103	.send(reset(&timeline_id, vec![item(1), item(2)]))
104	.unwrap();
105	for diffs in [
106	vec![
107	TimelineDiff::PushBack { value: item(3) },
108	TimelineDiff::Remove { index: 7 },
109	],
110	vec![TimelineDiff::PushBack { value: item(4) }],
111	] {
112	harness
113	.inject
114	.send(Event::TimelineChanged {
115	timeline_id: timeline_id.clone(),
116	diffs,
117	})
118	.unwrap();
119	}
120	next_matching(&mut updates, \|change\| {
121	timeline_change(change, &timeline_id, \|change\| {
122	matches!(change, TimelineChange::Diverged)
123	})
124	})
125	.await;
126	let snapshot = harness.mirror.snapshot();
127	let timeline = &snapshot.timelines[&timeline_id];
128	assert!(timeline.diverged);
129	assert_eq!(timeline.items, [item(1), item(2)].into_iter().collect());
130
131	harness
132	.inject
133	.send(reset(&timeline_id, vec![item(5)]))
134	.unwrap();
135	let update = next_matching(&mut updates, \|change\| {
136	timeline_change(change, &timeline_id, \|_\| true)
137	})
138	.await;
139	assert!(
140	timeline_change(&update.change, &timeline_id, \|change\| {
141	matches!(change, TimelineChange::Reset { .. })
142	}),
143	"a diverged timeline took {update:?}"
144	);
145	let snapshot = harness.mirror.snapshot();
146	let timeline = &snapshot.timelines[&timeline_id];
147	assert!(!timeline.diverged);
148	assert_eq!(timeline.items, [item(5)].into_iter().collect());
149	}
150
151	#[tokio::test]
152	async fn room_and_thread_timelines_stay_apart() {
153	let harness = start(EngineConfig::default(), 256).await;
154	let room = TimelineId::from(harness.room_id.clone());
155	let thread = TimelineId::thread(
156	harness.room_id.clone(),
157	event_id!("$root:localhost").to_owned(),
158	);
159	let mut updates = harness.mirror.subscribe();
160	harness.inject.send(reset(&room, vec![item(1)])).unwrap();
161	harness.inject.send(reset(&thread, vec![item(2)])).unwrap();
162	harness
163	.inject
164	.send(Event::TimelineChanged {
165	timeline_id: thread.clone(),
166	diffs: vec![TimelineDiff::PushBack { value: item(3) }],
167	})
168	.unwrap();
169	next_matching(&mut updates, \|change\| {
170	timeline_change(change, &thread, \|change\| {
171	matches!(change, TimelineChange::Changed { .. })
172	})
173	})
174	.await;
175	let snapshot = harness.mirror.snapshot();
176	assert_eq!(
177	snapshot.timelines[&room].items,
178	[item(1)].into_iter().collect()
179	);
180	assert_eq!(
181	snapshot.timelines[&thread].items,
182	[item(2), item(3)].into_iter().collect()
183	);
184	}
185
186	#[tokio::test]
187	async fn late_subscriber_gets_state_then_every_update_after_it() {
188	let harness = start(EngineConfig::default(), 1024).await;
189	let transaction_id = TransactionId::new();
190	let progress = \|uploaded_bytes\| Event::MediaUploadProgress {
191	room_id: harness.room_id.clone(),
192	transaction_id: transaction_id.clone(),
193	uploaded_bytes,
194	total_bytes: 100,
195	complete: false,
196	};
197	for uploaded_bytes in 1..=50 {
198	harness.inject.send(progress(uploaded_bytes)).unwrap();
199	}
200	let mut updates = harness.mirror.subscribe();
201	for uploaded_bytes in 51..=100 {
202	harness.inject.send(progress(uploaded_bytes)).unwrap();
203	}
204
205	let first = updates.next().await.unwrap();
206	let StateChange::State(state) = &first.change else {
207	panic!("a subscription opened with {first:?}");
208	};
209	let mut seen = state
210	.uploads
211	.get(&transaction_id)
212	.map_or(0, \|upload\| upload.uploaded_bytes);
213	let mut sequence = first.sequence;
214	while seen < 100 {
215	let update = next_matching(&mut updates, \|_\| true).await;
216	assert_eq!(update.sequence, sequence + 1);
217	sequence = update.sequence;
218	if let StateChange::Upload { upload, .. } = update.change {
219	assert_eq!(upload.uploaded_bytes, seen + 1);
220	seen += 1;
221	}
222	}
223	}
224
225	#[tokio::test]
226	async fn lagging_subscriber_starts_over_from_a_fresh_state() {
227	let harness = start(EngineConfig::default(), 4).await;
228	let transaction_id = TransactionId::new();
229	let mut updates = harness.mirror.subscribe();
230	assert!(matches!(
231	updates.next().await.unwrap().change,
232	StateChange::State(_)
233	));
234	for uploaded_bytes in 1..=20 {
235	harness
236	.inject
237	.send(Event::MediaUploadProgress {
238	room_id: harness.room_id.clone(),
239	transaction_id: transaction_id.clone(),
240	uploaded_bytes,
241	total_bytes: 20,
242	complete: uploaded_bytes == 20,
243	})
244	.unwrap();
245	}
246	tokio::time::timeout(Duration::from_secs(5), async {
247	while !harness
248	.mirror
249	.snapshot()
250	.uploads
251	.get(&transaction_id)
252	.is_some_and(\|upload\| upload.complete)
253	{
254	tokio::task::yield_now().await;
255	}
256	})
257	.await
258	.expect("the mirror never caught up");
259
260	let update = updates.next().await.unwrap();
261	let StateChange::State(state) = &update.change else {
262	panic!("a lagged subscription carried on with {update:?}");
263	};
264	assert!(state.uploads[&transaction_id].complete);
265	}
266
267	#[tokio::test]
268	async fn resync_does_not_deadlock_on_full_queues() {
269	let config = EngineConfig::default()
270	.with_command_capacity(1)
271	.with_event_capacity(1);
272	let harness = start(config, 1024).await;
273	let timeline_id = TimelineId::from(harness.room_id.clone());
274	let mut updates = harness.mirror.subscribe();
275	harness
276	.handle
277	.open_room(harness.room_id.clone())
278	.await
279	.unwrap();
280	next_matching(&mut updates, \|change\| {
281	timeline_change(change, &timeline_id, \|change\| {
282	matches!(change, TimelineChange::Reset { .. })
283	})
284	})
285	.await;
286
287	let flood = tokio::spawn({
288	let handle = harness.handle.clone();
289	let timeline_id = timeline_id.clone();
290	async move {
291	loop {
292	let command = Command::Timeline(TimelineCommand::Paginate {
293	timeline_id: timeline_id.clone(),
294	limit: 1,
295	});
296	if handle.send(command).await.is_err() {
297	return;
298	}
299	}
300	}
301	});
302	for _ in 0..10 {
303	harness
304	.inject
305	.send(Event::TimelineChanged {
306	timeline_id: timeline_id.clone(),
307	diffs: vec![TimelineDiff::Remove { index: 1000 }],
308	})
309	.unwrap();
310	next_matching(&mut updates, \|change\| {
311	timeline_change(change, &timeline_id, \|change\| {
312	matches!(change, TimelineChange::Diverged)
313	})
314	})
315	.await;
316	next_matching(&mut updates, \|change\| {
317	timeline_change(change, &timeline_id, \|change\| {
318	matches!(change, TimelineChange::Reset { .. })
319	})
320	})
321	.await;
322	}
323	flood.abort();
324	}
325
326	#[tokio::test]
327	async fn closing_a_timeline_evicts_it() {
328	let harness = start(EngineConfig::default(), 256).await;
329	let timeline_id = TimelineId::from(harness.room_id.clone());
330	let mut updates = harness.mirror.subscribe();
331	harness
332	.handle
333	.open_room(harness.room_id.clone())
334	.await
335	.unwrap();
336	harness
337	.handle
338	.close_room(harness.room_id.clone())
339	.await
340	.unwrap();
341	next_matching(&mut updates, \|change\| {
342	timeline_change(change, &timeline_id, \|change\| {
343	matches!(change, TimelineChange::Closed)
344	})
345	})
346	.await;
347	assert!(
348	!harness
349	.mirror
350	.snapshot()
351	.timelines
352	.contains_key(&timeline_id)
353	);
354	}
355
356	#[tokio::test]
357	async fn an_upload_ends_on_finished() {
358	let harness = start(EngineConfig::default(), 256).await;
359	let transaction_id = TransactionId::new();
360	let mut updates = harness.mirror.subscribe();
361	harness
362	.inject
363	.send(Event::MediaUploadProgress {
364	room_id: harness.room_id.clone(),
365	transaction_id: transaction_id.clone(),
366	uploaded_bytes: 5,
367	total_bytes: 10,
368	complete: false,
369	})
370	.unwrap();
371	next_matching(&mut updates, \|change\| {
372	matches!(change, StateChange::Upload { .. })
373	})
374	.await;
375	assert!(
376	harness
377	.mirror
378	.snapshot()
379	.uploads
380	.contains_key(&transaction_id)
381	);
382	harness
383	.inject
384	.send(Event::MediaUploadFinished {
385	room_id: harness.room_id.clone(),
386	transaction_id: transaction_id.clone(),
387	outcome: UploadOutcome::Cancelled,
388	})
389	.unwrap();
390	next_matching(&mut updates, \|change\| {
391	matches!(change, StateChange::UploadFinished { .. })
392	})
393	.await;
394	assert!(harness.mirror.snapshot().uploads.is_empty());
395	}